Patrons d’implementació

• Estructura del projecte
  • Patró: module-as-script (python -m)
    • El problema
    • La solució
  • Patró: path resolution
    • El problema
    • La solució
  • Patró: Makefile
    • El problema
    • La solució
    • Error típic
  • Patró: logging
    • El problema
    • La solució
    • On van els logs
    • Error típic
• Configuració i dades
  • Patró: configuration management
    • El problema
    • La solució
    • Decideix abans d’implementar
    • Error típic
  • Patró: ETL per a ML
    • El problema
    • Decideix abans d’implementar
    • Estructura bàsica
    • Error típic
• Servir prediccions
  • Patró: error handling
    • El problema
    • Els codis HTTP que necessites
    • Estructura recomanada
    • Cas concret: si falta un valor
  • Patró: dependency injection
    • El problema
    • La solució
• On trobar la resta

Aquest document recull receptes per a problemes concrets que trobaràs en portar un model a producció. Cada patró parteix d’un problema concret, en proposa una solució i, sovint, assenyala l’error més habitual. No és codi per copiar: llegeix el patró quan trobis el problema.

Els temes amb un capítol propi no es repeteixen aquí: la taula del final indica on trobar-los.

Estructura del projecte

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í?

Que funcioni o no depèn del directori des d’on l’executes i de la variable PYTHONPATH. Els imports entre mòduls del projecte fallen de maneres diferents a cada cas.

La solució

Organitza el codi en un paquet (app/) i executa cada script com a mòdul:

python -m app.train      # Sempre des de l'arrel del projecte
python -m app.pipeline

Cada mòdul té una funció main() i el bloc habitual:

# app/train.py
def main():
    ...

if __name__ == "__main__":
    main()

Així el mateix fitxer es pot executar (python -m app.train), importar (from app.train import main) i provar des d’un test.

Patró: path resolution

El problema

model = joblib.load("models/model.pkl")          # Depèn del directori de treball
model = joblib.load("/home/joan/model.pkl")      # Només funciona en una màquina

La solució

Calcula les rutes a partir de l’arrel del projecte, que es pot deduir de la ubicació del mateix fitxer:

# app/config.py
from pathlib import Path

PROJECT_ROOT = Path(__file__).parent.parent

def resolve_path(path: str) -> Path:
    """Resol una ruta relativa al projecte, sigui on sigui que s'hagi arrencat el procés."""
    return PROJECT_ROOT / path

La configuració guarda rutes relatives (models/model_v1.pkl) i el codi les resol. Funciona igual des de qualsevol directori i dins del contenidor.

Patró: Makefile

El problema

Cada persona de l’equip executa les coses a la seva manera, amb flags diferents, i el servidor de CI/CD repeteix les mateixes ordres en un altre script.

La solució

Un Makefile amb un target per a cada operació:

setup:  ## Crea l'entorn virtual i instal·la les dependències
	python3 -m venv .venv && .venv/bin/pip install -r requirements.txt

test:   ## Executa els tests
	.venv/bin/pytest tests/ -v

train:  ## Entrena una versió nova del model
	.venv/bin/python -m app.train

dev:    ## Arrenca l'API amb recàrrega automàtica
	.venv/bin/uvicorn app.api:app --reload --port 8080

Tothom fa servir les mateixes ordres (make test, make train), i si make health funciona a l’ordinador, funcionarà igual al servidor. Si el pipeline de CI/CD crida aquests mateixos targets, el que proves a l’ordinador és el que s’executa al servidor (vegeu Les mateixes ordres a tot arreu). Un Makefile no és l’única opció: un directori scripts/ o les ordres d’un pyproject.toml fan la mateixa funció.

Error típic

“Faig servir el mateix nom de target per a dues coses segons l’entorn.”

Un target fa sempre el mateix. Si cal comportament diferent, que vingui de la configuració (.env), no del nom.

Patró: logging

El problema

Alguna cosa falla en producció i només tens uns quants print() sense data, sense nivell i sense saber de quin mòdul vénen.

La solució

Configura el logging una sola vegada, al punt d’entrada:

import logging
import sys

def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
    handlers = [logging.StreamHandler(sys.stdout)]
    if log_file:  # només per als scripts batch que s'executen a mà
        handlers.append(logging.FileHandler(log_file))
    logging.basicConfig(
        level=getattr(logging, level.upper()),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
        handlers=handlers,
    )

I a cada mòdul, un logger amb el seu nom:

logger = logging.getLogger(__name__)

logger.info("Model loaded: version=%s", version)
logger.error("Prediction failed", exc_info=True)  # inclou la traça de l'error

On van els logs

• El servei (l’API) escriu només a la sortida estàndard. Docker la recull i la consultes amb docker compose logs api. Un fitxer dins del contenidor es perdria en recrear-lo (vegeu Logs).
• Els scripts batch que executes tu (make train, make pipeline) poden escriure també a un fitxer a logs/, perquè acaben i el fitxer queda a la teva màquina per revisar-lo.

Error típic

“Faig servir print() per depurar.”

Fes servir logger.debug(). En producció el nivell és INFO i aquests missatges desapareixen sense tocar el codi.

Configuració i dades

Patró: configuration management

El problema

model = joblib.load("/home/joan/project/models/model.pkl")
API_KEY = "sk-secret123"

Què passa quan un company clona el projecte, quan s’executa al servidor amb altres rutes, o quan algú llegeix la clau al repositori?

La solució

Separa el codi (el què) de la configuració (el on i el com). La configuració ve de variables d’entorn, normalment des d’un fitxer .env, i pydantic-settings la valida en arrencar:

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    MODEL_PATH: str              # obligatòria: sense valor per defecte
    METADATA_PATH: str           # obligatòria: sense valor per defecte
    API_KEY: str | None = None   # secret: ve de l'entorn, mai del repositori
    PREDICTIONS_LOG_PATH: str = "data/predictions.jsonl"
    LOG_LEVEL: str = "INFO"

settings = Settings()

I a qualsevol lloc del codi:

from app.config import settings

model = joblib.load(settings.MODEL_PATH)

Decideix abans d’implementar

Una variable obligatòria sense valor per defecte és una decisió deliberada: si falta, l’aplicació falla en arrencar, no a la primera petició. Els secrets no van mai al repositori: a la teva màquina, en un fitxer .env ignorat per git; en producció, al gestor de secrets de la plataforma (vegeu Configuració per entorn).

Error típic

“Tinc un config.py, però també faig os.getenv() en altres llocs.”

Centralitza tota la configuració a Settings. Un sol lloc per saber què es pot configurar.

Patró: ETL per a ML

El problema

Les dades no arriben netes. Cal obtenir-les (d’un fitxer, una base de dades, una API), transformar-les (validar, netejar, crear features) i guardar-les en un format adequat per entrenar. Aquest procés s’anomena ETL (Extract, Transform, Load).

Decideix abans d’implementar

• Extract: d’on vénen les dades? Caben a la memòria?
• Transform: quines files són invàlides i què en fas? Quins valors absents imputes aquí i quins deixes per al Pipeline del model?
• Load: en quin format? Parquet, per conservar els tipus (vegeu Dades per a producció).

Estructura bàsica

def run_pipeline(input_path: Path, output_dir: Path) -> None:
    df = pd.read_csv(input_path)                       # Extract
    df = clean(df)                                     # Transform
    train, val, prod = create_splits(df)               # 60/20/20
    for name, split in [("training_set", train),       # Load
                        ("validation_set", val),
                        ("production_set", prod)]:
        split.to_parquet(output_dir / f"{name}.parquet", index=False)

Tres principis:

• Idempotent: executar-lo dues vegades dona el mateix resultat. Fixa la llavor aleatòria dels splits (random_state=42).
• Registra cada pas: quantes files entren i quantes surten de cada transformació.
• Una fila errònia no atura tot el procés: es descarta i es registra.

Error típic

“Faig les transformacions al notebook i copio les dades a mà.”

El pipeline ha de ser un script executable (python -m app.pipeline). Si el dataset canvia, només cal tornar-lo a executar.

Servir prediccions

Patró: error handling

El problema

En producció les coses fallen, i la diferència entre un servei fet de pressa i un de professional és com comunica els errors:

• Fet de pressa: 500 Internal Server Error.
• Professional: un codi 4xx amb el missatge “odometer: ha de ser més gran o igual que 0”.

Els codis HTTP que necessites

FastAPI retorna 422 automàticament quan l’entrada no compleix l’esquema de Pydantic. Per a les validacions pròpies, hi ha APIs que fan servir 400 i d’altres que reutilitzen 422. Les dues opcions són vàlides; el que importa és triar-ne una i documentar-la, perquè el client sàpiga què esperar.

Estructura recomanada

from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    # Un missatge clar per al client, sense detalls interns
    fields = ", ".join(str(e["loc"][-1]) + ": " + e["msg"] for e in exc.errors())
    return JSONResponse(status_code=422,
                        content={"detail": {"code": "validation_error", "message": fields}})

@app.post("/predict")
def predict(car: CarFeatures):
    if "model" not in state:
        raise HTTPException(503, detail={"code": "model_not_loaded",
                                         "message": "Model not loaded"})
    try:
        price = float(state["model"].predict(pd.DataFrame([car.model_dump()]))[0])
    except Exception:
        logger.error("Prediction failed", exc_info=True)  # els detalls van al log
        raise HTTPException(500, detail={"code": "internal_error",
                                         "message": "Internal server error"})
    return {"price": price}

La regla és: detalls al log per a tu, missatges clars i sense dades internes per a l’usuari. Una resposta d’error no ha de mostrar mai rutes del servidor ni traces de Python. Un format fix ({"code", "message"}) permet al client reaccionar pel code sense analitzar el text.

Cas concret: si falta un valor

Una petició arriba sense un camp. Tens tres opcions, però no totes existeixen a les dues fases del model:

Dues idees:

• A producció, la decisió es redueix a rebutjar o imputar. Descartar només existeix en entrenar.
• La imputació lliga les dues fases. El valor amb què omples el buit (la mediana, la categoria més freqüent) es calcula amb les dades d’entrenament i es reutilitza en producció. Si és dins del Pipeline del model (un SimpleImputer), això passa sol (vegeu Model i preprocessament junts). Amb una sola petició no pots calcular cap mitjana.

Rebutjar o imputar? Pensa en el cost de l’error. Un camp poc important es pot imputar sense gaire risc. Si falta un camp clau per a la predicció, sovint és millor rebutjar la petició que retornar una predicció en què ningú no s’hauria de refiar.

Patró: dependency injection

El problema

L’endpoint de predicció necessita el model, un client per a un servei extern (un LLM, una base de dades) i la configuració. Si cada endpoint els crea pel seu compte, no es poden substituir als tests i el codi barreja lògica i infraestructura.

La solució

Crea el recurs un cop, en arrencar, i fes que l’endpoint el demani amb Depends:

from fastapi import Depends

@asynccontextmanager
async def lifespan(app: FastAPI):
    state["llm"] = OpenAILLMClient(settings.LLM_BASE_URL, settings.LLM_API_KEY)
    yield

def get_llm() -> LLMClient:
    return state["llm"]

@app.post("/predict")
def predict(listing: CarListing, llm: LLMClient = Depends(get_llm)):
    features = llm.extract(listing.description)
    ...

Als tests, substitueixes la dependència sense tocar l’API:

app.dependency_overrides[get_llm] = lambda: FakeLLMClient()

Avantatge clau: el test fa servir el mateix camí de codi que producció, només amb un client fals. Ho veiem en context a Substituir una dependència externa.

On trobar la resta