9  Despliegue de modelos como servicios

mlops
despliegue
api
rest
Autor/a

Wilder Ramírez Delgado

Fecha de publicación

12 de septiembre de 2026

Serie MLOps y Producción · de la ingeniería de datos al monitoreo.

Módulo IV — Despliegue y producción

Hasta aquí el artefacto vive en disco. Un usuario de negocio no abre un notebook ni llama a joblib.load. Llama a un servicio: envía un caso, espera una decisión. Este artículo enseña ese cambio de forma: de objeto Python a contrato HTTP.

No necesitamos levantar un puerto para entender el contrato. Entrenamos, persistimos (sesión 5 / 8) y simulamos request/respuesta como lo haría FastAPI. El código de FastAPI queda escrito para cuando el entorno lo tenga; la lección no depende de un servidor abierto.

Objetivos de aprendizaje

  1. Explicar cliente-servidor aplicado a un modelo.
  2. Definir un contrato REST de inferencia (esquema, status, cuerpo).
  3. Cargar el pipeline una vez y servir sin reentrenar.
  4. Distinguir 200, 422 y 500, y rechazar un request inválido.
  5. Documentar versión del contrato y latencia de una predicción.

Qué no cubre esta sesión

  • Empaquetar en Docker (sesión 10).
  • CI/CD (sesión 11).
  • Monitoreo de drift (sesión 12).

Ruta de trabajo

  1. De artefacto a servicio.
  2. REST sin mitología.
  3. Entrenar y persistir lo que se servirá.
  4. El servicio en proceso: validar, predecir, responder.
  5. FastAPI como especificación.
  6. Batch frente a online, latencia, versión.
  7. Taller.
# Entorno: artefacto de servicio y reloj de latencia.

import json
import time
from pathlib import Path

import joblib
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

np.random.seed(42)
RUTA = Path("artefactos_sesion9")
RUTA.mkdir(exist_ok=True)
print("Entorno listo.")
print("El servidor cargará el joblib UNA vez. Cada request solo transforma y predice.")
Entorno listo.
El servidor cargará el joblib UNA vez. Cada request solo transforma y predice.

1) De artefacto a servicio

En una frase: desplegar es poner \(f(T(x))\) detrás de una interfaz que otros procesos puedan llamar sin conocer Python.

Tres formas, de menos a más sistema

Forma Quién ejecuta \(f\) Límite
Notebook / script El autor No hay usuarios concurrentes
Lote (batch) Un job nocturno No hay respuesta en segundos
Servicio Un proceso que escucha Hay contrato, errores, carga

Un servicio no es “el modelo en la nube”. Es un proceso de larga vida que carga una vez el .joblib (el de production de la sesión 8) y, por cada request, valida, aplica \(T\) y \(f\), y responde.

Reentrenar en cada request negaría las sesiones 5 y 6: caro, no reproducible y distinto en cada llamada.

Cliente y servidor

El cliente (app, otro API, un curl) no tiene el modelo. Tiene un caso: ingreso, segmento. El servidor tiene el artefacto y la regla de validación. Se hablan por HTTP: el cliente no importa un .py; importa una URL y un JSON.

Si el modelo cambia y el contrato JSON se mantiene, el cliente no se reescribe. Si segmento pasa a llamarse tipo_cliente y nadie avisa, el cliente “sigue igual” y el servidor se llena de 422.

2) REST, sin mitología

En una frase: REST, aquí, es un estilo: recursos, verbos HTTP y códigos de estado que significan algo para el cliente.

Para inferencia suele bastar:

  • POST /predict — el cuerpo es el caso; la respuesta es la decisión.
  • GET /salud — el proceso está vivo y el artefacto se pudo cargar.
Código Significado en inferencia
200 Caso válido, hay predicción
422 El JSON no cumple el esquema
500 Falló el servidor (artefacto corrupto, bug)

422 no es un capricho. Es la frontera del servicio: no se predice sobre un caso que \(T\) no sabe transformar. Un 200 con null es peor: finge éxito.

El cuerpo de entrada es el esquema crudo de la sesión 5, no las dummies. El cliente no escala. El pipeline sí.

3) Entrenar una vez, persistir lo que se servirá

# Dataset mínimo y pipeline del contrato de las sesiones 4–5.
rng = np.random.default_rng(42)
n = 360
ingreso = rng.normal(4_500_000, 1_100_000, n).clip(800_000, 10_000_000)
segmento = rng.choice(["retail", "pyme", "corporativo"], n, p=[0.55, 0.30, 0.15])
logit = -2.0 + 1.4 * ((ingreso - ingreso.mean()) / ingreso.std()) + np.where(segmento == "corporativo", 1.0, 0.0)
y = (rng.random(n) < 1 / (1 + np.exp(-logit))).astype(int)
df = pd.DataFrame({"ingreso": ingreso, "segmento": segmento, "y": y})

pipe = Pipeline(
    [
        (
            "pre",
            ColumnTransformer(
                [
                    ("num", StandardScaler(), ["ingreso"]),
                    ("cat", OneHotEncoder(handle_unknown="ignore"), ["segmento"]),
                ]
            ),
        ),
        ("clf", LogisticRegression(max_iter=400)),
    ]
)
pipe.fit(df[["ingreso", "segmento"]], df["y"])
ruta_modelo = RUTA / "modelo.joblib"
joblib.dump(pipe, ruta_modelo)
print("Artefacto:", ruta_modelo.resolve())
print("El servidor lo cargará una vez, no en cada request.")
Artefacto: C:\Users\wilde\Dropbox\bit-blog\mlops-produccion\09-despliegue-modelos-servicios\artefactos_sesion9\modelo.joblib
El servidor lo cargará una vez, no en cada request.

Ese archivo debería ser una copia de registry/production (sesión 8). Aquí se fabrica en la misma carpeta para que el notebook sea autónomo. En un sistema, esta celda no existiría: existiría un load.

4) El servicio como funciones: validar y predecir

# Contrato de entrada y salida, independiente de FastAPI.
ESQUEMA = {"columnas": ["ingreso", "segmento"], "segmentos": ["retail", "pyme", "corporativo"]}
MODELO = joblib.load(ruta_modelo)


def validar_caso(caso: dict) -> list[str]:
    errores = []
    if not isinstance(caso, dict):
        return ["el cuerpo no es un objeto"]
    for c in ESQUEMA["columnas"]:
        if c not in caso:
            errores.append(f"falta {c}")
    if errores:
        return errores
    try:
        float(caso["ingreso"])
    except (TypeError, ValueError):
        errores.append("ingreso no numérico")
    if caso.get("segmento") not in ESQUEMA["segmentos"]:
        errores.append(f"segmento desconocido: {caso.get('segmento')}")
    return errores


def predecir(caso: dict) -> dict:
    frame = pd.DataFrame([{"ingreso": float(caso["ingreso"]), "segmento": caso["segmento"]}])
    proba = float(MODELO.predict_proba(frame)[0, 1])
    etiqueta = int(proba >= 0.5)
    return {"etiqueta": etiqueta, "probabilidad": round(proba, 4), "umbral": 0.5}

print("Servicio en proceso. Modelo cargado:", type(MODELO).__name__)
Servicio en proceso. Modelo cargado: Pipeline

MODELO es una variable de arranque. Si estuviera dentro de predecir, cada request releería el disco: latencia y, peor, la tentación de reentrenar.

5) El manejador HTTP simulado

# Un request es un dict; la respuesta tiene status y cuerpo.

def handle_predict(caso):
    t0 = time.perf_counter()
    errores = validar_caso(caso)
    if errores:
        return {"status": 422, "cuerpo": {"errores": errores}, "ms": (time.perf_counter()-t0)*1000}
    try:
        cuerpo = predecir(caso)
        return {"status": 200, "cuerpo": cuerpo, "ms": (time.perf_counter()-t0)*1000}
    except Exception as exc:
        return {"status": 500, "cuerpo": {"error": str(exc)}, "ms": (time.perf_counter()-t0)*1000}


def handle_salud():
    return {"status": 200, "cuerpo": {"ok": True, "modelo_cargado": MODELO is not None}}

print(handle_salud())
print(handle_predict({"ingreso": 7_200_000, "segmento": "corporativo"}))
{'status': 200, 'cuerpo': {'ok': True, 'modelo_cargado': True}}
{'status': 200, 'cuerpo': {'etiqueta': 1, 'probabilidad': 0.8945, 'umbral': 0.5}, 'ms': 7.457500000100481}

6) Tres requests que el cliente debe poder distinguir

# 200, 422 por falta, 422 por dominio.
casos = [
    ("válido", {"ingreso": 3.2e6, "segmento": "pyme"}),
    ("falta campo", {"ingreso": 3.2e6}),
    ("segmento nuevo", {"ingreso": 3.2e6, "segmento": "gobierno"}),
    ("tipo incorrecto", {"ingreso": "mucho", "segmento": "retail"}),
]
filas = []
for nombre, caso in casos:
    r = handle_predict(caso)
    filas.append({"caso": nombre, "status": r["status"], "cuerpo": r["cuerpo"], "ms": round(r["ms"], 3)})
print(pd.DataFrame(filas).to_string(index=False))
           caso  status                                                 cuerpo    ms
         válido     200 {'etiqueta': 0, 'probabilidad': 0.0407, 'umbral': 0.5} 6.379
    falta campo     422                        {'errores': ['falta segmento']} 0.004
 segmento nuevo     422        {'errores': ['segmento desconocido: gobierno']} 0.004
tipo incorrecto     422                   {'errores': ['ingreso no numérico']} 0.010

El nivel gobierno aquí se rechaza en validación (422). El OneHotEncoder con handle_unknown="ignore" podría predecir. Esa es una decisión de contrato: ¿el servicio acepta niveles nuevos como “otros” o los rechaza? Hay que declararla. Esta sesión rechaza: un nivel no catalogado es un 422, no un 200 silencioso.

7) FastAPI como especificación, no como dependencia de la lección

El siguiente bloque es el mapa del servicio real. Si el entorno no tiene FastAPI, no impide entender el contrato: ya lo ejecutamos arriba.

# Especificación FastAPI (documental). No se arranca un puerto.

especificacion = '''
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

class Caso(BaseModel):
    ingreso: float = Field(..., gt=0)
    segmento: str

app = FastAPI(title="prioridad", version="1.0.0")

@app.get("/salud")
def salud():
    return {"ok": True}

@app.post("/predict")
def predict(caso: Caso):
    errores = validar_caso(caso.model_dump())
    if errores:
        raise HTTPException(status_code=422, detail=errores)
    return predecir(caso.model_dump())
'''
print(especificacion)
print("Pydantic cubre parte del 422 (tipos). validar_caso cubre el dominio (segmento).")

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

class Caso(BaseModel):
    ingreso: float = Field(..., gt=0)
    segmento: str

app = FastAPI(title="prioridad", version="1.0.0")

@app.get("/salud")
def salud():
    return {"ok": True}

@app.post("/predict")
def predict(caso: Caso):
    errores = validar_caso(caso.model_dump())
    if errores:
        raise HTTPException(status_code=422, detail=errores)
    return predecir(caso.model_dump())

Pydantic cubre parte del 422 (tipos). validar_caso cubre el dominio (segmento).

8) Batch frente a online

Modo Entrada Latencia esperada Riesgo
Online 1 JSON decenas de ms Pico de QPS
Batch 10⁴ filas minutos Mezclar \(T\) “del batch” con la de producción

Un endpoint /predict_batch que recibe 10⁴ filas sigue usando el mismo MODELO. No recalcula \(\mu\). Si alguien escala el batch a mano, volvió a la sesión 5.

# Batch: mismo contrato, varias filas, misma T congelada.
lote = pd.DataFrame(
    [
        {"ingreso": 2.1e6, "segmento": "retail"},
        {"ingreso": 6.8e6, "segmento": "corporativo"},
        {"ingreso": 4.0e6, "segmento": "pyme"},
    ]
)
pred_lote = MODELO.predict(lote)
proba_lote = MODELO.predict_proba(lote)[:, 1]
print(lote.assign(etiqueta=pred_lote, probabilidad=np.round(proba_lote, 3)).to_string(index=False))
  ingreso    segmento  etiqueta  probabilidad
2100000.0      retail         0         0.010
6800000.0 corporativo         1         0.844
4000000.0        pyme         0         0.095

9) Latencia: lo que el cliente siente

# 200 requests sintéticos: distribución de ms del manejador.
lat = [handle_predict({"ingreso": 4.5e6, "segmento": "pyme"})["ms"] for _ in range(200)]
print(f"p50={np.median(lat):.3f} ms | p95={np.percentile(lat, 95):.3f} ms | max={np.max(lat):.3f} ms")
fig, ax = plt.subplots(figsize=(7.2, 3.6))
ax.hist(lat, bins=20, color="slategray", edgecolor="white")
ax.set_xlabel("ms")
ax.set_title("Latencia del predict en proceso (sin red)")
plt.tight_layout()
plt.show()
print("En un servidor real se suma la red. Si p95 ya es alto aquí, el modelo (no HTTP) es el cuello.")
p50=6.241 ms | p95=12.938 ms | max=20.305 ms

En un servidor real se suma la red. Si p95 ya es alto aquí, el modelo (no HTTP) es el cuello.

10) Versión del contrato

El modelo puede cambiar (otro run_id de la sesión 8) sin cambiar el JSON. O al revés: el JSON gana un campo canal y los clientes viejos rompen. Por eso se versiona el contrato (/v1/predict) aparte del artefacto.

# Cabecera de versión: el cliente puede registrarla.
CONTRATO = {"nombre": "prioridad", "version_api": "v1", "run_id_esperado": "production"}
print(json.dumps({**CONTRATO, **handle_salud()["cuerpo"]}, indent=2))
{
  "nombre": "prioridad",
  "version_api": "v1",
  "run_id_esperado": "production",
  "ok": true,
  "modelo_cargado": true
}

11) Idempotencia de GET y no idempotencia de POST

GET /salud no cambia estado: se puede repetir. POST /predict en este servicio tampoco cambia el modelo (no hay aprendizaje online). Aun así se usa POST porque el cuerpo es un caso, no un recurso cacheable. Si mañana el servicio escribiera la predicción en una bitácora de auditoría, el POST dejaría de ser inocuo: habría que diseñar ids de request.

12) Cargar mal el artefacto es un 500 de arranque, no de request

# Arranque: si el archivo no existe, el servicio no debe 'escuchar' como si nada.
ruta_falsa = RUTA / "no_existe.joblib"
try:
    joblib.load(ruta_falsa)
    print("no debería")
except Exception as exc:
    print("Fallo de arranque (equivalente a no pasar /salud):")
    print(" ", type(exc).__name__, str(exc)[:140])
Fallo de arranque (equivalente a no pasar /salud):
  FileNotFoundError [Errno 2] No such file or directory: 'artefactos_sesion9\\no_existe.joblib'

Un orquestador (sesión 7) no pondría tráfico encima de un proceso cuyo /salud ya falló. Un balanceador tampoco. /salud existe para eso, no para “hacer ping”.

13) El cliente no reimplementa \(T\)

# Anti-patrón: el cliente escala a mano y rompe el contrato.
caso = {"ingreso": 4.5e6, "segmento": "pyme"}
ok = handle_predict(caso)["cuerpo"]
# Un cliente 'listo' envía z-score en lugar de pesos.
mal = handle_predict({"ingreso": 0.12, "segmento": "pyme"})["cuerpo"]
print("Contrato crudo :", ok)
print("Ingreso ya 'escalado' por el cliente:", mal)
print("Misma etiqueta?:", ok["etiqueta"] == mal["etiqueta"])
print("Si no coinciden, el cliente inventó otra T.")
Contrato crudo : {'etiqueta': 0, 'probabilidad': 0.1551, 'umbral': 0.5}
Ingreso ya 'escalado' por el cliente: {'etiqueta': 0, 'probabilidad': 0.0012, 'umbral': 0.5}
Misma etiqueta?: True
Si no coinciden, el cliente inventó otra T.

14) Relación con 8 y 10

El servicio carga un puntero: production. El registry decide cuál. Docker (sesión 10) empaqueta este proceso. CI (sesión 11) evita promover un artefacto que no pasa /salud y un predict de humo.

14b) Umbral: ¿en el modelo o en el servicio?

El umbral 0.5 está hoy en predecir. Moverlo al servicio permite cambiar la decisión de negocio sin reentrenar. Moverlo sin versionar el contrato hace que dos clientes “iguales” no lo sean. Se declara en la respuesta (umbral: 0.5) para que el auditor no adivine.

# Mismo score, dos umbrales: la etiqueta es política, no solo estadística.
caso = {"ingreso": 5.5e6, "segmento": "pyme"}
proba = predecir(caso)["probabilidad"]
print("probabilidad:", proba)
print("umbral 0.50 →", int(proba >= 0.50))
print("umbral 0.35 →", int(proba >= 0.35))
print("Cambiar umbral en el servicio exige versionar la política, no solo el joblib.")
probabilidad: 0.3617
umbral 0.50 → 0
umbral 0.35 → 1
Cambiar umbral en el servicio exige versionar la política, no solo el joblib.

14c) Autenticación, en una frase

Un endpoint de predicción sin autenticación es una puerta. Esta sesión no implementa OAuth: deja dicho que el contrato HTTP incluye quién puede llamar. El 401/403 no se improvisa el día del incidente.

14d) Observabilidad mínima del request

Además de la latencia, un servicio serio registra (sin PII innecesaria): status, versión, run_id, un id de request. La sesión 12 usará esa bitácora para drift. Sin ella, el monitoreo no tiene de dónde beber.

# Bitácora de servicio (JSONL), sin guardar el ingreso crudo si no hace falta.
log_path = RUTA / "requests.jsonl"
registros = []
for nombre, caso in casos:
    r = handle_predict(caso)
    rec = {"caso": nombre, "status": r["status"], "ms": round(r["ms"], 4), "api": "v1"}
    registros.append(rec)
with log_path.open("w", encoding="utf-8") as fh:
    for rec in registros:
        fh.write(json.dumps(rec) + "\n")
print(log_path.read_text(encoding="utf-8"))
{"caso": "v\u00e1lido", "status": 200, "ms": 7.7701, "api": "v1"}
{"caso": "falta campo", "status": 422, "ms": 0.006, "api": "v1"}
{"caso": "segmento nuevo", "status": 422, "ms": 0.0046, "api": "v1"}
{"caso": "tipo incorrecto", "status": 422, "ms": 0.0133, "api": "v1"}

14e) CORS y clientes de navegador

Si el cliente es una página web en otro origen, el navegador bloqueará el POST salvo que el servidor declare CORS. No es un detalle de frontend: es parte del contrato de exposición. Un curl no lo sufre; un alumno que ‘ya le funciona en Postman’ sí.

14f) Payload máximo

Un /predict_batch sin tope es un vector de denegación de servicio. Se declara un máximo de filas (por ejemplo 500) y se responde 413 o 422 si se excede.

MAX_LOTE = 500
print("Política: lotes >", MAX_LOTE, "se rechazan.")
print("Este laboratorio no abre el puerto; la política debe quedar en el código del servicio.")
Política: lotes > 500 se rechazan.
Este laboratorio no abre el puerto; la política debe quedar en el código del servicio.

14g) Predicción vs explicación

Devolver coeficientes en cada 200 mezcla inferencia con la sesión 13. Si se ofrece /explain, es otro recurso, otro costo, otra versión. /predict se mantiene delgado.

14h) Prueba de humo

# Tres asserts que un CI (sesión 11) debería correr contra el servicio en proceso.
r_ok = handle_predict({"ingreso": 4.5e6, "segmento": "retail"})
r_bad = handle_predict({"segmento": "retail"})
r_salud = handle_salud()
assert r_salud["status"] == 200
assert r_ok["status"] == 200 and "etiqueta" in r_ok["cuerpo"]
assert r_bad["status"] == 422
print("Humo en proceso: /salud 200, predict válido 200, predict cojo 422.")
Humo en proceso: /salud 200, predict válido 200, predict cojo 422.

14i) El mismo caso, dos cargas del joblib

# Releer de disco no debe cambiar la predicción (contrato de la sesión 5).
otro = joblib.load(ruta_modelo)
caso_df = pd.DataFrame([{"ingreso": 4.5e6, "segmento": "retail"}])
print("¿Idéntico al recargar?:",
      np.array_equal(MODELO.predict(caso_df), otro.predict(caso_df)))
¿Idéntico al recargar?: True

14j) Mapa de rutas

rutas = pd.DataFrame(
    [
        {"metodo": "GET", "ruta": "/salud", "status_ok": 200},
        {"metodo": "POST", "ruta": "/v1/predict", "status_ok": 200},
        {"metodo": "POST", "ruta": "/v1/predict", "status_ok": 422},
        {"metodo": "GET", "ruta": "/v1/predict", "status_ok": 405},
    ]
)
print(rutas.to_string(index=False))
print("405: el verbo importa. Un GET a /predict no es un atajo.")
metodo        ruta  status_ok
   GET      /salud        200
  POST /v1/predict        200
  POST /v1/predict        422
   GET /v1/predict        405
405: el verbo importa. Un GET a /predict no es un atajo.

14k) Ejemplo de curl (texto)

print('curl -s -X POST http://localhost:8000/v1/predict \\')
print('  -H "Content-Type: application/json" \\')
print('  -d "{\"ingreso\": 4500000, \"segmento\": \"pyme\"}"')
curl -s -X POST http://localhost:8000/v1/predict \
  -H "Content-Type: application/json" \
  -d "{"ingreso": 4500000, "segmento": "pyme"}"

15) Errores frecuentes

  1. Reentrenar en cada POST.
  2. Devolver 200 con null cuando faltan campos.
  3. Dejar que el cliente escale.
  4. Cargar registry/staging en producción “porque era el último”.
  5. No tener /salud.
  6. Cambiar el JSON sin subir /v2.
  7. Confundir este servicio con el contenedor (sesión 10).

16) Checklist

  1. ¿El joblib se carga al arrancar?
  2. ¿El 422 cubre esquema y dominio?
  3. ¿El 200 siempre trae etiqueta y, si se acordó, probabilidad?
  4. ¿/salud refleja el arranque, no un ping vacío?
  5. ¿El cliente envía crudo?
  6. ¿La versión del contrato está escrita?

17) Conclusiones

Desplegar es traducir \(f(T(x))\) a un contrato que otro proceso puede cumplir. El código de FastAPI es un mapa. El examen de esta sesión es otro: un 422 a tiempo, un 200 honesto, un modelo que no se reentrena por request, y un puntero que debería venir del registry. La sesión 10 meterá este proceso en una caja que se puede llevar a otra máquina.

18) Taller práctico

A. Añada visitas al esquema. Haga fallar un request viejo (sin el campo) con 422.

B. Implemente handle_predict_batch que rechace el lote si una fila es inválida (todo o nada).

C. Simule un 500: apunte MODELO a None y capture el status. Restaure.

D. Compare latencia de 1 fila vs 200 filas en un solo predict de sklearn.

E. Escriba el curl equivalente (como texto) de un POST /v1/predict válido.

print("Modelo:", ruta_modelo.exists(), "| log:", log_path.exists())
print("Ejemplo 200:", handle_predict({"ingreso": 4.5e6, "segmento": "retail"}))
print("Taller A–E. Si el request inválido devuelve 200, el contrato no existe.")
Modelo: True | log: True
Ejemplo 200: {'status': 200, 'cuerpo': {'etiqueta': 0, 'probabilidad': 0.1332, 'umbral': 0.5}, 'ms': 7.968300000129602}
Taller A–E. Si el request inválido devuelve 200, el contrato no existe.

💬 ¿Te sirvió?

Deja en los comentarios una duda o un caso donde aplicarías esto — respondo todos. Sígueme para no perderte el próximo artículo de la serie y comparte con alguien que esté aprendiendo arquitectura de IA.

👉 El código completo está disponible para ejecutar directamente.