FastAPI per microservizi ML con Pydantic v2

Perché FastAPI per Microservizi ML?

Nel 2026, deploying ML models in production richiede tre cose: velocità (latency < 200ms), validazione rigorosa (no garbage input), e resilienza (fallback quando il modello è down). FastAPI + Pydantic v2 + Python 3.12 risponde a questi requisiti in modo nativo.

A differenza di Flask (legacy, lento, no type hints), FastAPI offre:

• Async/await nativo: gestisci centinaia di richieste simultanee su una sola CPU core.
• Pydantic v2 integrata: validazione schema-first, serialization automatica, error messages precise.
• OpenAPI docs auto-generate: Swagger UI gratis, no docstring a mano.
• Type hints: code intelligence negli IDE, catch bugs prima del runtime.
• Dependency injection: iniettare modelli, cache, logger, settings in modo dichiarativo.

Nel portfolio di Federico Calò, il microservizio ML FastAPI on VPS Hetzner serve 40+ endpoint RAG con latenza media 85ms (p99 < 350ms) con zero downtime, gestendo 2M richieste/mese.

Pydantic v2: Modelli e Validatori

Definire Modelli Request/Response

In Pydantic v2, un modello è una class che eredita da BaseModel. La magia: type hints → validazione automatica.

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

class EmbeddingRequest(BaseModel):
    """Richiesta di embedding: testo → vettore."""
    text: str = Field(..., min_length=1, max_length=2000, description="Testo da embeddare")
    model: str = Field(default="sentence-transformers/all-MiniLM-L6-v2")
    normalize: bool = Field(default=True, description="Normalizzare il vettore?")

    @field_validator('text')
    @classmethod
    def text_not_only_spaces(cls, v):
        if not v.strip():
            raise ValueError("Testo non può essere solo spazi")
        return v.strip()

class EmbeddingResponse(BaseModel):
    """Risposta embedding."""
    embedding: list[float] = Field(..., description="Vettore embedding (dim 384)")
    model: str
    dims: int
    computation_time_ms: float = Field(..., ge=0, description="Tempo elaborazione")
  

Note:

• Field(...) = obbligatorio, Field(default=...) = facoltativo.
• field_validator = validatore custom eseguito dopo il parsing.
• ge=0 = greater or equal (schema validation).

SettingsModel per Configurazione

Pydantic v2 introduce BaseSettings per caricare variabili d'ambiente con type hints e validazione:

from pydantic_settings import BaseSettings
from pydantic import Field

class MLServiceSettings(BaseSettings):
    """Configurazione del microservizio ML."""
    ollama_url: str = Field(default="http://localhost:11434", description="URL Ollama locale")
    ollama_model: str = Field(default="llama3.1:8b")
    embedding_model: str = Field(default="sentence-transformers/all-MiniLM-L6-v2")
    max_context_length: int = Field(default=2048, ge=256)
    health_check_interval: float = Field(default=30.0, gt=0)
    timeout_seconds: float = Field(default=60.0, gt=0)

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = MLServiceSettings()
  

Carica da .env o da variabili di sistema. Se OLLAMA_URL non è valido (es. non-URL), Pydantic solleva ValidationError al startup — fail-fast.

Async/Await per I/O Non-Bloccante

FastAPI è built on async. Ogni endpoint deve essere async def per sfruttare I/O multiplexing via Uvicorn (ASGI server).

import httpx
import asyncio

class OllamaClient:
    """Client asincrono verso Ollama on-prem."""

    def __init__(self, base_url: str, timeout: float = 60.0):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self._client: Optional[httpx.AsyncClient] = None

    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._client:
            await self._client.aclose()

    async def generate(self, prompt: str, model: str = "llama3.1:8b") -> str:
        """Genera testo via Ollama (streaming)."""
        if not self._client:
            raise RuntimeError("Client non inizializzato")

        response = await self._client.post(
            f"{self.base_url}/api/generate",
            json={"model": model, "prompt": prompt, "stream": False}
        )
        response.raise_for_status()
        return response.json()["response"]
  

Con async with OllamaClient(...) as client: text = await client.generate(...), il thread Uvicorn non blocca mentre aspetta Ollama — può servire altre richieste in parallelo.

Lifespan: Caricamento Modelli al Boot

La sezione più critica. Con lifespan, carichi il modello una sola volta al startup (invece che per ogni richiesta). FastAPI v0.93+ usa @asynccontextmanager:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# Stato globale (singleton)
class ModelCache:
    embedding_model = None
    ollama_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # STARTUP: carica modelli
    print("[STARTUP] Caricamento modelli...")

    try:
        # Carica embedding model in-memory
        from sentence_transformers import SentenceTransformer
        ModelCache.embedding_model = SentenceTransformer(
            "sentence-transformers/all-MiniLM-L6-v2",
            device="cpu"  # o "cuda" se GPU disponibile
        )
        print("[STARTUP] Embedding model caricato")

        # Crea client Ollama
        ModelCache.ollama_client = OllamaClient(
            base_url=settings.ollama_url,
            timeout=settings.timeout_seconds
        )
        await ModelCache.ollama_client.__aenter__()
        print("[STARTUP] Ollama client connesso")

    except Exception as e:
        print(f"[ERROR] Startup fallito: {e}")
        raise

    yield  # App gira qui

    # SHUTDOWN: cleanup
    print("[SHUTDOWN] Chiusura risorse...")
    if ModelCache.ollama_client:
        await ModelCache.ollama_client.__aexit__(None, None, None)
    print("[SHUTDOWN] Cleanup completo")

app = FastAPI(
    title="ML Microservice",
    version="1.0.0",
    lifespan=lifespan
)
  

Il lifespan garantisce:

• Modelli caricati UNA volta al boot (non per richiesta).
• Se il caricamento fallisce, l'app non avvia (fail-fast).
• Shutdown orderly: chiudi connessioni, libera memoria.

Dependency Injection

FastAPI's Depends() inietta dipendenze negli endpoint in modo dichiarativo:

from fastapi import FastAPI, Depends

def get_embedding_model():
    """Dependency: ritorna il modello embedding caricato."""
    if ModelCache.embedding_model is None:
        raise RuntimeError("Embedding model non caricato")
    return ModelCache.embedding_model

def get_ollama_client():
    """Dependency: ritorna il client Ollama."""
    if ModelCache.ollama_client is None:
        raise RuntimeError("Ollama client non disponibile")
    return ModelCache.ollama_client

@app.post("/embed", response_model=EmbeddingResponse)
async def embed(
    request: EmbeddingRequest,
    model = Depends(get_embedding_model),
    ollama_client = Depends(get_ollama_client)
):
    """Embedding endpoint con dipendenze iniettate."""
    import time
    start = time.perf_counter()

    # Usa model e ollama_client
    embedding = model.encode(request.text).tolist()

    elapsed_ms = (time.perf_counter() - start) * 1000

    return EmbeddingResponse(
        embedding=embedding,
        model=request.model,
        dims=len(embedding),
        computation_time_ms=elapsed_ms
    )
  

Vantaggi: testabilità (mock le dependencies), riusabilità (stessa dependency in più endpoint), e error propagation (se il modello è None, FastAPI risponde 500 con messaggio chiaro).

Validazione Request/Response Automatica

Pydantic v2 valida automaticamente:

# Richiesta valida
POST /embed
{"text": "Hello world", "normalize": true}
→ 200 OK

# Richiesta invalida (testo troppo lungo)
POST /embed
{"text": "x" * 2001}
→ 422 Unprocessable Entity
{
  "detail": [
    {
      "type": "string_too_long",
      "loc": ["body", "text"],
      "msg": "String should have at most 2000 characters",
      "input": "xxxx..."
    }
  ]
}
  

Niente try/catch a mano per validare input — Pydantic lo fa automaticamente.

Error Handling Strutturato

Definisci eccezioni custom e un global exception handler:

from fastapi import HTTPException, status

class OllamaUnavailableError(Exception):
    """Ollama non disponibile."""
    pass

class ModelInferenceError(Exception):
    """Errore durante l'inferenza."""
    pass

@app.exception_handler(OllamaUnavailableError)
async def handle_ollama_unavailable(request, exc):
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={
            "error": "ollama_unavailable",
            "message": "Ollama AI provider momentaneamente non disponibile",
            "retry_after_seconds": 30
        }
    )

@app.exception_handler(ModelInferenceError)
async def handle_inference_error(request, exc):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "error": "inference_failed",
            "message": str(exc),
            "request_id": request.headers.get("X-Request-ID", "unknown")
        }
    )
  

Health Check e Graceful Degradation

Un health check endpoint che monitora lo stato dei servizi dipendenti:

from enum import Enum
from datetime import datetime

class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

class HealthCheck(BaseModel):
    status: HealthStatus
    timestamp: str
    services: dict[str, dict]
    version: str = "1.0.0"

@app.get("/health", response_model=HealthCheck)
async def health_check():
    """Health check con stato di tutti i servizi dipendenti."""
    import time

    services = {}
    overall_status = HealthStatus.HEALTHY

    # Check embedding model
    if ModelCache.embedding_model is None:
        services["embedding_model"] = {"status": "down"}
        overall_status = HealthStatus.UNHEALTHY
    else:
        services["embedding_model"] = {"status": "up"}

    # Check Ollama
    if ModelCache.ollama_client is None:
        services["ollama"] = {"status": "down", "message": "Client non inizializzato"}
        overall_status = HealthStatus.UNHEALTHY
    else:
        try:
            start = time.perf_counter()
            response = await ModelCache.ollama_client._client.get(
                f"{ModelCache.ollama_client.base_url}/api/tags",
                timeout=5.0
            )
            elapsed_ms = (time.perf_counter() - start) * 1000

            if response.status_code == 200:
                services["ollama"] = {
                    "status": "up",
                    "latency_ms": round(elapsed_ms, 2)
                }
            else:
                services["ollama"] = {
                    "status": "degraded",
                    "http_code": response.status_code
                }
                overall_status = HealthStatus.DEGRADED

        except Exception as e:
            services["ollama"] = {"status": "down", "error": str(e)}
            overall_status = HealthStatus.UNHEALTHY

    return HealthCheck(
        status=overall_status,
        timestamp=datetime.utcnow().isoformat(),
        services=services
    )
  

Graceful Degradation: Fallback Deterministico

Se Ollama non risponde, il servizio ritorna un risultato credibile (non crash):

class ChatRequest(BaseModel):
    query: str = Field(..., max_length=1000)
    system_prompt: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    model: str
    ollama_available: bool
    latency_ms: float

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """Chat endpoint con graceful degradation."""
    import time
    start = time.perf_counter()

    ollama_available = False
    response_text = ""

    try:
        # Prova Ollama
        async with OllamaClient(settings.ollama_url, timeout=10.0) as client:
            response_text = await client.generate(
                prompt=request.query,
                model=settings.ollama_model
            )
            ollama_available = True

    except asyncio.TimeoutError:
        # Ollama timeout: fallback
        response_text = f"[FALLBACK] Ollama non rispondeva nel timeout. Query: {request.query[:100]}..."
        ollama_available = False

    except Exception as e:
        # Qualunque altro errore
        response_text = f"[FALLBACK] Errore servizio: {type(e).__name__}"
        ollama_available = False

    elapsed_ms = (time.perf_counter() - start) * 1000

    return ChatResponse(
        response=response_text,
        model=settings.ollama_model,
        ollama_available=ollama_available,
        latency_ms=round(elapsed_ms, 2)
    )
  

Esempio RAG Completo: Embedding + Retrieval + Generation

Un microservizio RAG realistico che combina tutto:

import numpy as np
from typing import List

class DocumentStore:
    """In-memory document store con embedding."""

    def __init__(self):
        self.docs: List[dict] = []
        self.embeddings: List[np.ndarray] = []

    def add_document(self, doc_id: str, text: str, embedding: np.ndarray):
        """Aggiungi documento con embedding."""
        self.docs.append({"id": doc_id, "text": text})
        self.embeddings.append(embedding)

    async def search(self, query_embedding: np.ndarray, top_k: int = 3) -> List[dict]:
        """Retrieval: ritorna i top-k documenti più simili."""
        if not self.embeddings:
            return []

        scores = [
            np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb) + 1e-8)
            for emb in self.embeddings
        ]

        top_indices = np.argsort(scores)[-top_k:][::-1]
        return [
            {"doc": self.docs[i], "score": float(scores[i])}
            for i in top_indices
            if scores[i] > 0.3  # Solo documenti correlati
        ]

# Istanza globale
document_store = DocumentStore()

class RAGRequest(BaseModel):
    query: str = Field(..., max_length=500)
    top_k: int = Field(default=3, ge=1, le=10)
    context_window: int = Field(default=2000, ge=256, le=4096)

class RAGResponse(BaseModel):
    query: str
    retrieved_docs: List[dict]
    generated_response: str
    model: str
    retrieval_latency_ms: float
    generation_latency_ms: float

@app.post("/rag", response_model=RAGResponse)
async def rag_search_and_generate(
    request: RAGRequest,
    embedding_model = Depends(get_embedding_model),
    ollama_client = Depends(get_ollama_client)
):
    """RAG: retrieval + generation."""
    import time

    # 1. Embed query
    retrieval_start = time.perf_counter()
    query_embedding = embedding_model.encode(request.query)

    # 2. Retrieve documenti correlati
    retrieved = await document_store.search(query_embedding, top_k=request.top_k)
    retrieval_ms = (time.perf_counter() - retrieval_start) * 1000

    # 3. Build context
    context = "\n".join([
        f"[Doc {i}] {doc['doc']['text'][:200]}"
        for i, doc in enumerate(retrieved)
    ])

    # 4. Generate response via Ollama
    generation_start = time.perf_counter()
    rag_prompt = f"""
Context:
{context}

Question: {request.query}

Answer:"""

    try:
        generated = await ollama_client.generate(prompt=rag_prompt)
    except Exception as e:
        generated = f"[Errore generazione]: {str(e)}"

    generation_ms = (time.perf_counter() - generation_start) * 1000

    return RAGResponse(
        query=request.query,
        retrieved_docs=retrieved,
        generated_response=generated,
        model=settings.ollama_model,
        retrieval_latency_ms=round(retrieval_ms, 2),
        generation_latency_ms=round(generation_ms, 2)
    )
  

Testing Asincrono con pytest-asyncio

import pytest
from fastapi.testclient import TestClient

@pytest.mark.asyncio
async def test_health_check():
    """Test health check quando modelli sono caricati."""
    async with client as c:
        response = await c.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] in ["healthy", "degraded"]
        assert "services" in data

@pytest.mark.asyncio
async def test_embed_validation():
    """Test validazione Pydantic su request invalida."""
    async with client as c:
        response = await c.post("/embed", json={"text": ""})
        assert response.status_code == 422  # Unprocessable Entity
        errors = response.json()["detail"]
        assert any("String should have at most" in e.get("msg", "") for e in errors)
  

Performance Optimization

Nell'ambiente di produzione di Federico Calò (VPS Hetzner CCX13, 4 CPU core):

• Latenza embedding: 5-8ms (per testo < 100 word).
• Latenza RAG retrieval: 2-3ms (cosine similarity su 10k doc).
• Latenza Ollama generation: 150-300ms (model llama3.1:8b, p99).
• Throughput: 40+ richieste concorrenti senza stress.

Bottleneck: il modello LLM (Ollama) — 85% della latenza totale. Soluzione: cache generazioni common queries (Redis o in-memory).

FAQ

Q: Posso usare FastAPI con un database PostgreSQL?

A: Sì, tramite SQLAlchemy async + Alembic. Nel portfolio, tutti i tool generatori usano SQLAlchemy con connection pooling asincrono per query rapide (< 50ms).

Q: Come gestisco rate limiting?

A: Usa slowapi (libreria FastAPI) o implementa middleware custom con Redis. Nel microservizio ML: 120 req/min per client IP (via header X-Forwarded-For).

Q: Pydantic v2 è backward-compatible con v1?

A: No, breaking changes significativi. Migration path: Config diventa model_config, parse_obj diventa model_validate. Vedi docs ufficiali per lista completa.

Q: Come monitoro il microservizio in produzione?

A: Esponi metrica Prometheus via prometheus-client: latenza richieste, errori per tipo, uso memoria. Nel portfolio: Prometheus scrape ogni 30s + Grafana dashboard per visualizzazione.

Q: Devo usare Docker?

A: Consigliato ma non obbligatorio. Nel portfolio: Docker Compose per dev, systemd unit per VPS produzione. Dockerfile standard: Python 3.12 slim + pip install + CMD uvicorn.