Patrons d’implementació per MLOps

Com usar aquest document
Part 1: Estructura i tooling
Part 2: Gestió de configuració i dades
- Patró: Configuration Management
- Patró: ETL per ML
Part 3: Servir prediccions
- Patró: Online vs Batch Prediction
- Patró: Error Handling
Part 4: Persistència i base de dades
- Patró: SQLAlchemy amb FastAPI
- Patró: Dependency Injection
Part 5: Testing
- Patró: Testing per ML Systems
- Patró: Test Database Fixtures
  - El problema
  - La solució: BD temporal en memòria
Part 6: Qualitat i operacions
Resum: Preguntes per cada patró

Com usar aquest document

Aquest no és un manual per copiar i enganxar. Cada patró resol un problema concret que trobareu quan porteu un model de ML a producció. Abans de mirar el codi, enteneu:

Quin problema resol - Per què necessiteu aquest patró?
Quan aplicar-lo - En quines situacions és útil?
Quines decisions heu de prendre - Què heu de pensar abans d’implementar?

Quan trobeu un problema al vostre projecte, busqueu el patró corresponent aquí.

Part 1: Estructura i tooling

Patrons per organitzar el projecte. Aquestes són les bases que cal establir abans de construir funcionalitats.

Patró: Module-as-Script (python -m)

El problema


python scripts/train.py        # Funciona?
./scripts/train.py             # I això?
cd scripts && python train.py  # I des d'aquí?

Depèn del directori, del PYTHONPATH, de moltes coses.

La solució


python -m app.train     # Sempre funciona
python -m app.pipeline  # Des de qualsevol lloc

Com implementar-ho


# app/train.py
def main():
    # Lògica principal

if __name__ == "__main__":
    main()

Ara podeu:

Executar: python -m app.train
Importar: from app.train import main
Testejar: main() directament

Patró: Path Resolution

El problema


model = joblib.load("models/model.pkl")        # Depèn del working directory
model = joblib.load("/home/joan/model.pkl")    # No portable

La solució


# config.py
from pathlib import Path

PROJECT_ROOT = Path(__file__).parent.parent

class Settings(BaseSettings):
    model_path: str = "models/model.pkl"

    def get_model_path(self) -> Path:
        return PROJECT_ROOT / self.model_path

Ara funciona des de qualsevol directori i a Docker.

Patró: Makefile com a Interfície

El problema

Cada desenvolupador executa comandes diferents
Difícil recordar tots els flags
CI/CD duplica scripts

La solució

El Makefile és el vostre contracte d’interfície:


setup:
    python -m venv venv && pip install -r requirements.txt

test:
    pytest tests/ -v

docker-up:
    docker compose up -d

health:
    curl http://localhost:8000/health

Tothom interactua amb el projecte igual: make test, make docker-up, make health.

Avantatge clau

Si make health funciona localment, funcionarà al servidor CI (mateixes comandes).

Patró: Structured Logging

El problema

Quan alguna cosa falla en producció, com sabeu què ha passat? La primera instíncta és usar print(), però té limitacions clares:

	`print()`	`import logging`
Format	Cada llamada té el seu propi format	Un seol format definit una vegada, aplicat a tota la sortida
Sortides	Sempre sols stdout	Podeu enviar a consola, fitxer, servei extern… sense canviar el codi
Filtrat	No podeu desactivar-lo sense esborrar línies	Controleu el nivell (DEBUG, INFO, ERROR…) per entorn
Ubicació	Cal cercar on va fer el `print`	Cada missatge porta automàticament el nom del module i la línia

La clau és que amb logging configureu una vegada i tot el codi del projecte n’aprofita automàticament.

Nivells de log

Nivell	Quan usar-lo
DEBUG	Detalls per desenvolupament
INFO	Confirmació que les coses funcionen
WARNING	Alguna cosa inesperada però no crítica
ERROR	Error que impedeix una operació
CRITICAL	Error greu, el sistema pot fallar

Estructura bàsica

Primer, configureu el root logger amb els dos handlers — un per consola i un per fitxer. Feu-ho una vegada, normalment al punt d’entrada del projecte (ex: main() o app/config.py):


import logging

def setup_logging(log_level: str = "INFO", log_file: str = "app.log"):
    handlers = [
        logging.StreamHandler(),                          # Consola (stdout)
        logging.FileHandler(log_file),                    # Fitxer
    ]

    logging.basicConfig(
        level=getattr(logging, log_level.upper()),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        handlers=handlers,
    )

Després, a cada module on necessiteu logar, creeu un logger amb el nom del module:


import logging
logger = logging.getLogger(__name__)

logger.info("Model carregat: version=%s", version)
logger.error("Predicció fallida", exc_info=True)  # Inclou traceback

Penseu-hi

Què logar: Startup, cada predicció (sense dades sensibles), errors
On enviar: Console (per Docker), fitxer (per persistència)
Quin nivell per entorn: DEBUG en dev, INFO/WARNING en prod

Error típic

“Uso print() per debugging”

Useu logger.debug(). En producció podeu desactivar debug sense tocar el codi.

Part 2: Gestió de configuració i dades

Un projecte real té configuració (paths, ports, credencials) i dades que cal processar. Com organitzar-ho?

Patró: Configuration Management

El problema

Mireu aquest codi:


model = joblib.load("/home/joan/projecte/models/model.pkl")
DB_PASSWORD = "secret123"

Què passa quan:

Un altre membre de l’equip clona el projecte?
Voleu executar en un servidor amb paths diferents?
Algú veu la contrasenya al repositori git?

La solució: Variables d’entorn

Separeu què (el codi) de on/com (la configuració):


.env (fitxer)  →  pydantic-settings (validació)  →  settings (objecte Python)

Penseu-hi

Abans d’implementar, classifiqueu la vostra configuració:

Configuració	Obligatòria?	Té default?	És secreta?
MODEL_PATH	Sí	No	No
API_PORT	No	Sí (8000)	No
DB_PASSWORD	Sí	No	Sí

Les obligatòries sense default fallaran si no existeixen - això és bo, detecteu errors ràpid.

Estructura bàsica


# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_path: str                    # Obligatòria
    api_port: int = 8000               # Amb default
    log_level: str = "INFO"            # Amb default

    class Config:
        env_file = ".env"

settings = Settings()

Ara a qualsevol lloc:


from config import settings
model = joblib.load(settings.model_path)

Error típic

“Tinc config.py però també faig os.environ.get() en altres llocs”

Centralitzeu tota la configuració. Un sol punt d’entrada.

Patró: ETL per ML

El problema

Les dades rarament arriben netes i llestes. Necessiteu:

Obtenir-les (d’un fitxer, base de dades, API…)
Transformar-les (validar, netejar, crear features…)
Guardar-les (en format adequat per entrenar)

Això és ETL: Extract, Transform, Load.

Penseu-hi

Per cada fase:

Extract:

D’on venen les dades?
Són massa grans per carregar a memòria d’un cop?

Transform:

Quines validacions necessiteu?
Què feu amb registres invàlids?
Quines transformacions apliqueu?

Load:

Quin format de sortida? (CSV llegible, Parquet eficient)
Guardeu metadades (quants registres, quan es va processar)?

Estructura bàsica


def run_pipeline(input_path: Path, output_dir: Path):
    # EXTRACT
    df = pd.read_csv(input_path)

    # TRANSFORM
    df = validar(df)
    df = netejar(df)
    df = crear_features(df)

    # SPLIT (específic per ML)
    train, val, test = dividir(df)

    # LOAD
    guardar(train, val, test, output_dir)

Principis importants

Idempotència: Executar dues vegades → mateix resultat
Logging: Registrar cada pas (quants registres entren/surten)
Error handling: Un registre erroni no ha de fer fallar tot

Error típic

“Faig les transformacions al notebook i copio les dades manualment”

El pipeline ha de ser un script executable. Si canvia el dataset, només cal tornar a executar.

Part 3: Servir prediccions

Quan el model està entrenat, heu de decidir com el fareu accessible. Aquesta decisió té impacte en tota l’arquitectura.

Patró: Online vs Batch Prediction

El problema

Teniu un model entrenat. Ara què? Hi ha dues estratègies fonamentalment diferents:

	Online	Batch
Metàfora	Un cambrer que serveix plats a demanda	Una cuina industrial que prepara 1000 menús
Quan s’executa	Quan arriba cada petició	Programat (cada hora, nit, setmana…)
Qui espera	L’usuari, davant la pantalla	Ningú - és un procés en background

Penseu-hi

Abans d’implementar, responeu aquestes preguntes:

L’usuari final espera una resposta immediata?
Necessiteu processar moltes dades d’un cop (milers de registres)?
La predicció és per a una acció immediata o per a un report?

Si l’usuari espera → Online. Si processeu en background → Batch.

Quan usar Online

Recomanacions mentre l’usuari navega
Detecció de frau en una transacció
Predicció de preu en una app

L’estructura bàsica és:


Petició HTTP → Validar input → Model.predict() → Retornar JSON

Quan usar Batch

Generar un report diari amb prediccions per tots els clients
Preprocessar un dataset gran per anàlisi
Calcular scores que es consultaran després (cache)

L’estructura bàsica és:


Llegir fitxer → Iterar per chunks → Guardar resultats

Error típic

“Faig un bucle que crida l’API per cada registre del CSV”

Això és el pitjor dels dos mons: la latència d’online amb el volum de batch. Si teniu molts registres, processeu-los en batch o envieu-los tots de cop a l’API.

Patró: Error Handling

El problema

En producció, les coses fallen. La diferència entre un sistema amateur i un professional és com comunica els errors.

Compareu:

Amateur: “500 Internal Server Error”
Professional: “El camp ‘age’ ha de ser un número positiu (rebut: -5)”

Els codis HTTP que heu de conèixer

Codi	Significat	Exemple
200	Tot bé	Predicció exitosa
400	Error de l’usuari	Dades invàlides
422	Dades mal formades	JSON incorrecte
500	Error del servidor	Bug al codi
503	Servei no disponible	Model no carregat

Penseu-hi

Quins errors poden passar? (Feu una llista)
Quins són culpa de l’usuari? (→ 4xx)
Quins són culpa del sistema? (→ 5xx)
Quin missatge és útil per l’usuari sense revelar detalls interns?

Estructura recomanada


@app.post("/predict")
def predict(features: Features):
    # 1. Validació de negoci
    if not es_valid(features):
        raise HTTPException(400, detail="Explicació clara del problema")

    try:
        # 2. Lògica principal
        prediction = model.predict(...)
        return {"prediction": prediction}

    except HTTPException:
        raise  # Re-llençar errors HTTP

    except Exception as e:
        # 3. Log intern (amb detalls), resposta genèrica (sense detalls)
        logger.error("Error", exc_info=True)
        raise HTTPException(500, detail="Error intern del servidor")

La clau és: logs detallats per vosaltres, missatges clars per l’usuari.

Part 4: Persistència i base de dades

Ara que l’API serveix prediccions i gestiona errors, necessiteu guardar-les per auditoria, debugging i monitoring.

Patró: SQLAlchemy amb FastAPI

El problema

Voleu guardar cada predicció per:

Auditoria: Qui va demanar què i quan
Debugging: Investigar prediccions estranyes
Monitoring: Estadístiques d’ús i latència

Penseu-hi

Quina base de dades?
- SQLite: Zero configuració, ideal per prototips
- PostgreSQL: Robust, per producció real
Què guardeu de cada predicció?
- Timestamp
- Input features (totes? les importants?)
- Output
- Versió del model
- Latència

Estructura bàsica

Primer, definiu què voleu guardar:


class PredictionLog(Base):
    __tablename__ = "predictions"

    id = Column(Integer, primary_key=True)
    timestamp = Column(DateTime, default=datetime.utcnow)

    # TODO: Decidiu quins camps d'input guardeu
    # ...

    prediction = Column(Float)
    model_version = Column(String)
    latency_ms = Column(Float)

Després, integreu amb l’API usant dependency injection.

Patró: Dependency Injection

El problema

L’endpoint de predicció necessita accés a:

El model
La base de dades
La configuració

Com ho organitzeu sense crear un embolic?

Sense DI (problemàtic)


@app.post("/predict")
def predict(features: Features):
    db = SessionLocal()       # Crear connexió
    model = load_model()      # Carregar model
    try:
        # ... lògica ...
    finally:
        db.close()            # Recordar tancar!

Problemes:

Difícil de testejar (sempre usa recursos reals)
Fàcil oblidar netejar recursos
Lògica barrejada amb infraestructura

Amb DI (recomanat)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/predict")
def predict(features: Features, db: Session = Depends(get_db)):
    # FastAPI gestiona la connexió automàticament
    # ...

Avantatge clau: Per testejar, podeu substituir get_db per una funció que retorna una BD de test.

Part 6: Qualitat i operacions

El testing valida que el codi és correcte. Ara, abans de desplegar, necessiteu també validar que el sistema complet està llest per producció.

Patró: Quality Gates

El problema

Un model serialitzat existeix (.pkl), però pot tenir accuracy de 0.3, features incompatibles, o un training que va fallar silenciosament. Com eviteu que arribi a producció?

Penseu-hi

Definiu els vostres criteris de desplegament abans d’implementar:

Quina mètrica és crítica pel vostre problema? (accuracy, F1, MAE…)
Quin és el llindar mínim acceptable?
Hi ha altres condicions? (features presents, latència màxima…)

Documenteu aquests criteris en un fitxer de configuració (config/deployment_criteria.yaml o similar).

Com implementar-ho

Principi clau: El model no es considera desplegable per defecte. Ha de passar una validació explícita. Només llavors es marca com deployment_ready: true a la metadata.

Al final del vostre script d’entrenament:


# Després d'entrenar i avaluar
metrics = {"accuracy": accuracy, "f1_score": f1}

# Carregar criteris
criteria = load_criteria("config/deployment_criteria.yaml")

# Decisió: passa o no passa?
deployment_ready = (
    metrics["accuracy"] >= criteria["min_accuracy"] and
    metrics["f1_score"] >= criteria["min_f1"]
)

# Guardar metadata amb el flag
metadata = {
    "version": version,
    "metrics": metrics,
    "deployment_ready": deployment_ready  # Resultat de la validació
}
save_metadata(metadata)

El hook pre-deploy.sh del CI/CD llegeix deployment_ready de la metadata. Si és False → el pipeline falla i el model no es desplega.


make test              # Tests passen?
python validate.py     # Model deployment_ready?
make docker-build      # Es pot construir?
make health            # Servei respon?

Si qualsevol passa falla → NO es desplega.

Error típic

“El meu training script sempre posa deployment_ready: true”

Si el flag és sempre True independentment de les mètriques, no esteu protegint res. El valor ha de ser el resultat d’una validació real contra criteris definits prèviament.

Patró: Health Check

El problema

Com sabeu si el servei està funcionant correctament?

Els load balancers necessiten saber si poden enviar tràfic
El monitoring necessita alertar si hi ha problemes
Després d’un deploy, voleu verificar que tot funciona

Penseu-hi

Què ha de verificar el health check?

Model carregat? (model is not None)
Base de dades accessible? (db.execute("SELECT 1"))
Espai en disc suficient?

Estructura bàsica


@app.get("/health")
def health_check():
    model_ok = model is not None
    db_ok = check_database()

    if model_ok and db_ok:
        return {"status": "healthy", "model_version": VERSION}
    else:
        raise HTTPException(503, detail="Service unhealthy")

Error típic

“El health check fa una predicció real per verificar el model”

Els health checks han de ser ràpids (< 1 segon). Verificar que el model està carregat és suficient.

Patró: Model Versioning

El problema

Teniu un model en producció. Entreno un de nou. Com gestioneu les versions?

El versionat de models (semantic versioning, metadata, estructura de directoris, symlinks, model registry) es tracta en detall al capítol de desplegament. El punt clau per a implementació és afegir el flag deployment_ready a la metadata:


models/
├── model_v1.0.0.pkl       # Model serialitzat
└── metadata_v1.0.0.json   # Informació sobre el model

El metadata ha d’incloure: versió, data d’entrenament, mètriques, noms de features, i el flag deployment_ready (generat pel Quality Gate).

Les versions anteriors del model guardades
Una manera de canviar quin model es carrega sense modificar codi
Un procés de reinici ràpid

La solució

Si el model es carrega des d’una variable d’entorn:


# config.py
class Settings(BaseSettings):
    model_path: str = "models/model_v1.2.0.pkl"

Fer rollback és:


# 1. Canviar .env
MODEL_PATH=models/model_v1.1.0.pkl

# 2. Reiniciar el servei
docker-compose restart api

Sense canviar codi, sense fer commit, sense CI/CD.

Què fa possible un rollback ràpid

Versions guardades: Mai sobreescriu el model anterior
Configuració externa: El path del model és una variable, no hardcoded
Metadata: Saps quina versió estàs servint i pots comparar

Error típic


# ❌ Sempre carrega "model.pkl"
model = joblib.load("models/model.pkl")

Quan fas un nou entrenament, sobreescrius el model anterior. Si alguna cosa falla, no pots tornar enrere.


# ✓ Versions explícites
model = joblib.load(settings.model_path)  # Controlat per .env

Resum: Preguntes per cada patró

Quan implementeu un patró, responeu primer:

Patró	Pregunta clau
Module-as-Script	Com executar scripts de manera consistent?
Path Resolution	Com referenciar fitxers independentment del working directory?
Makefile	Quines operacions ha de poder fer qualsevol membre de l’equip?
Logging	Què necessiteu saber quan alguna cosa falla?
Configuration	Quines configuracions són obligatòries vs opcionals?
ETL	D’on venen les dades i com les valideu?
Online vs Batch	L’usuari espera resposta immediata?
Error Handling	Quins errors poden passar i qui és responsable?
SQLAlchemy	Què guardeu de cada predicció?
Dependency Injection	Quins recursos necessita cada endpoint?
Testing	Què és determinista (testejar) vs estocàstic (no testejar)?
Quality Gates	Quins criteris ha de passar el codi abans de desplegar?
Health Check	Quins components ha de verificar?
Model Versioning	Quan incrementeu major/minor/patch?
Rollback	Què necessiteu tenir preparat abans que el model falli?

Keyboard shortcuts

apunts.jg5.dev