mlops · temario
5.1tema 1 de 5

FastAPI — Model Serving

Exponer un modelo de ML por REST: Pydantic, validación, async y Swagger.

Un modelo entrenado que vive en un notebook no produce valor real. Para que un modelo sea útil en producción necesita una interfaz: una API REST que permita a cualquier sistema consultarlo con datos de entrada y recibir una predicción. FastAPI es el estándar de facto en el ecosistema Python para este propósito porque combina velocidad (basado en Starlette y uvicorn), tipado estático y generación automática de documentación.

Pydantic es la columna vertebral de la validación en FastAPI. Cada request y cada response se define como una clase que hereda de BaseModel. Si el cliente envía un campo con el tipo incorrecto, FastAPI rechaza el request antes de que llegue a tu función de predicción con un error 422 claro y descriptivo. Esto elimina una clase entera de bugs: los errores de tipo en tiempo de ejecución dentro del modelo.

El decorador @app.post con el parámetro response_model cumple dos funciones críticas: valida que la respuesta que produces tiene exactamente la forma prometida, y genera la documentación en Swagger/OpenAPI automáticamente. Si tu función devuelve un campo extra que no está en el modelo, FastAPI lo filtra. Si te falta un campo obligatorio, falla en tiempo de desarrollo, no en producción.

FastAPI es asíncrono por diseño (async/await). Sin embargo, la inferencia de modelos como scikit-learn, XGBoost o PyTorch es código bloqueante (CPU-bound). Si llamas model.predict() dentro de una función async sin delegarla a un executor, bloqueas el event loop y tu API deja de responder a otras requests. La solución es usar asyncio.run_in_executor o directamente definir el endpoint como función síncrona regular (def en vez de async def) y dejar que FastAPI lo maneje en un threadpool.

La documentación interactiva en /docs (Swagger UI) y /redoc se genera sola a partir de tus modelos Pydantic y los decoradores de ruta. Esto no es un lujo: es contrato vivo. Si tu equipo de frontend o de data science necesita saber qué enviar y qué esperar, abren /docs y lo ven ejecutable. Arquitectura de productivización aplicada (SageMaker + FastAPI + Lambda) que ejemplifica este patrón en producción real: https://churn.benjacode.com

El ciclo de vida de la aplicación (lifespan) permite cargar el modelo una sola vez al arrancar el servidor en lugar de cargarlo en cada request. Cargar un modelo de 500 MB en cada llamada destruye la latencia. Con el context manager lifespan puedes hacer model = load_model() al startup y guardarlo en app.state, dejándolo disponible para todos los endpoints sin costo de inicialización por request.

structure.txt
# Install
pip install fastapi uvicorn pydantic

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pydantic import BaseModel, Field
import joblib

# Request / Response models
class PredictRequest(BaseModel):
    age: int = Field(..., ge=0, le=120)
    tenure: float = Field(..., ge=0.0)
    monthly_charges: float

class PredictResponse(BaseModel):
    churn_probability: float
    label: str

# Load model once at startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.model = joblib.load('model.pkl')
    yield

app = FastAPI(title='Churn API', lifespan=lifespan)

# Sync endpoint — FastAPI runs it in threadpool automatically
@app.post('/predict', response_model=PredictResponse)
def predict(req: PredictRequest):
    model = app.state.model
    features = [[req.age, req.tenure, req.monthly_charges]]
    prob = model.predict_proba(features)[0][1]
    return PredictResponse(
        churn_probability=round(prob, 4),
        label='churn' if prob > 0.5 else 'no_churn',
    )

# Health check
@app.get('/health')
def health():
    return {'status': 'ok'}

# Run
# uvicorn main:app --host 0.0.0.0 --port 8000

Debugging lab

Detecta y corrige el error en el código.

0/5 tests passing0%
  1. 5.1.5.1

    class PredictRequest(BaseModel): age: int tenure: float class PredictResponse(BaseModel): probability: str label: str @app.post('/predict') def predict(req: PredictRequest): prob = model.predict_proba([[req.age, req.tenure]])[0][1] return {'probability': prob, 'label': 'churn'}

  2. 5.1.5.2

    @app.post('/predict', response_model=PredictResponse) async def predict(req: PredictRequest): features = [[req.age, req.tenure]] prob = model.predict_proba(features)[0][1] return PredictResponse(probability=prob, label='churn')

  3. 5.1.5.3

    @asynccontextmanager async def lifespan(app: FastAPI): yield app = FastAPI(lifespan=lifespan) @app.post('/predict', response_model=PredictResponse) def predict(req: PredictRequest): model = joblib.load('model.pkl') prob = model.predict_proba([[req.age]])[0][1] return PredictResponse(probability=prob, label='churn')

  4. 5.1.5.4

    class PredictRequest(BaseModel): features: list @app.post('/predict') def predict(req: PredictRequest): result = model.predict(req.features) return result.tolist()

  5. 5.1.5.5

    @app.post('/predict', response_model=PredictResponse) def predict(req: PredictRequest): try: prob = app.state.model.predict_proba([[req.age]])[0][1] return PredictResponse(probability=prob, label='churn') except Exception as e: return {'error': str(e), 'traceback': traceback.format_exc()}