El Projecte | Sobre nosaltres | Contribuir | Donacions | Llicència
HOME Els meus cursos Finalitzats
◄ Anterior
◄

Un servei de predicció amb FastAPI

Següent ►
►

La decisió està presa: a més del batch setmanal per a màrqueting, CineClick necessita un servei online que, en el moment en què un client prem "cancel·lar subscripció", retorni la seva probabilitat d'abandonament en menys de 300 ms perquè producte decideixi si mostra una oferta. En aquesta lliçó construïm aquest servei amb FastAPI: definirem contractes d'entrada i de sortida amb Pydantic, carregarem el campió des del registry en arrencar, exposarem els endpoints /predir, /salut i /versio, i — complint el mandat de la lliçó 03-03 — importarem construir_features del paquet instal·lable en lloc de reescriure-la. Al final tindràs el servei corrent a la teva màquina amb uvicorn, provat des de la documentació interactiva i des de curl, i deixant als logs la llavor del monitoratge que arribarà al mòdul 6.

Contingut

  1. Per què FastAPI per servir models
  2. Estructura del servei al repo
  3. Esquemes Pydantic: el contracte d'entrada i sortida
  4. Carregar el model en arrencar: lifespan
  5. L'endpoint POST /predir
  6. Healthcheck i versió: GET /salut i GET /versio
  7. Provar el servei: uvicorn, /docs i curl
  8. Logging estructurat de cada predicció

Per què FastAPI per servir models

Hi ha moltes maneres d'exposar un model per HTTP (Flask, Django, un servidor fet a mà). FastAPI s'ha convertit en l'estàndard de facto per servir ML en Python per quatre raons concretes:

  • Tipatge amb Pydantic: defineixes l'entrada i la sortida com a classes Python amb tipus, i FastAPI valida automàticament cada petició contra aquest esquema. En un servei de ML això és or: la causa número u de prediccions absurdes en producció és una entrada malformada que el model s'empassa sense queixar-se (un pla inexistent, una antiguitat negativa). Amb Pydantic, la petició invàlida mor a la porta amb un error 422 explicatiu, abans de tocar el model.
  • Documentació OpenAPI automàtica: el servei publica a /docs una interfície interactiva generada a partir dels esquemes. L'equip de producte de CineClick pot veure i provar el contracte sense llegir ni una línia de Python.
  • Suport async natiu: FastAPI corre sobre un servidor ASGI (uvicorn) i pot atendre moltes connexions concurrents. Matís important per a ML: el predict de scikit-learn és codi síncron lligat a CPU, així que el declararem en un endpoint def normal (no async def) i FastAPI l'executarà en un threadpool sense bloquejar el bucle d'esdeveniments.
  • Rendiment i ergonomia: és del més ràpid de l'ecosistema Python i el codi resultant és curt i llegible — veuràs que el servei complet cap en dos fitxers.

Afegim les dependències al projecte (i les congelem al lock, com mana el mòdul 2):

# afegir a pyproject.toml a [project.dependencies]:
#   fastapi==0.115.5
#   uvicorn[standard]==0.32.1
pip-compile pyproject.toml -o requirements.lock   # regenerar el lock
pip install -e .                                   # reinstal·lar el paquet

Estructura del servei al repo

El servei viu dins del paquet cineclick_churn, no en un repo a part. Aquesta decisió és conseqüència directa de 03-03: el servei ha d'importar construir_features del mateix paquet que fa servir l'entrenament, i tenir-los junts garanteix que tots dos s'instal·len i es versionen alhora.

cineclick-churn/
├── src/cineclick_churn/
│   ├── data.py
│   ├── features.py          # font única de features (03-03)
│   ├── train.py
│   ├── evaluate.py
│   └── api/                 # NOU: el servei de predicció
│       ├── __init__.py
│       ├── main.py          # l'app FastAPI: lifespan + endpoints
│       └── schemas.py       # contractes Pydantic d'entrada/sortida
├── tests/
├── configs/
└── ...

Dos fitxers: schemas.py defineix què entra i surt; main.py defineix què se'n fa. Separar-los manté el contracte llegible per si sol.

Esquemes Pydantic: el contracte d'entrada i sortida

L'esquema d'entrada replica les columnes crues de clients_churn.csv (menys id_client, que viatja a part per traçabilitat, i abandonament, que és el que predim). Fixa't que les validacions codifiquen regles de domini, no només tipus:

# src/cineclick_churn/api/schemas.py
from typing import Literal
from pydantic import BaseModel, Field

class ClientEntrada(BaseModel):
    """Dades crues d'un client, tal com les té el frontend."""

    id_client: str = Field(..., min_length=1, description="Identificador del client")
    antiguitat_mesos: int = Field(..., ge=0, le=600,
                                  description="Mesos des de l'alta (>= 0)")
    hores_setmana: float = Field(..., ge=0, le=168,
                                 description="Hores de visionat aquesta setmana")
    tiquets_suport: int = Field(..., ge=0,
                                description="Tiquets de suport oberts")
    pla: Literal["basic", "estandard", "premium"]
    metode_pagament: Literal["targeta", "domiciliacio", "paypal"]
    descompte_actiu: bool

class PrediccioSortida(BaseModel):
    """Resposta del servei: probabilitat, decisió i traçabilitat."""

    id_client: str
    probabilitat_abandonament: float = Field(..., ge=0, le=1)
    prediccio: bool                      # True = es prediu abandonament
    llindar: float                       # llindar aplicat en aquesta resposta
    versio_model: str                    # p. ex. "2" — la versió del registry

Desglossament de les decisions:

  • ge=0 a antiguitat_mesos i tiquets_suport talla en sec les entrades negatives — que, recorda, trencarien silenciosament la feature ratio_tickets (el clip(lower=1) protegeix contra divisions estranyes, però una antiguitat de -3 és una dada corrupta que no ha d'arribar tan lluny).
  • le=168 a hores_setmana: una setmana té 168 hores; més que això és un bug de l'emissor, no un client molt fan.
  • Literal[...] a pla i metode_pagament restringeix als valors amb què el model es va entrenar. Si demà negoci llança el pla "familiar", l'API retornarà 422 en lloc de deixar que el model al·lucini sobre una categoria que no ha vist mai — i aquest 422 és l'alarma primerenca que cal reentrenar i actualitzar el contracte.
  • PrediccioSortida retorna no només la probabilitat sinó el llindar aplicat i la versió del model: cada resposta és autoexplicativa i auditable. Quan al mòdul 6 investiguem una deriva, saber quina versió va respondre cada petició serà imprescindible.

Carregar el model en arrencar: lifespan

Regla d'or del serving: el model es carrega una vegada, en arrencar el procés — mai dins de l'endpoint. Carregar des del registry costa segons; fer-ho per petició destruiria qualsevol pressupost de latència. FastAPI ofereix el patró lifespan per executar codi a l'arrencada i a l'aturada:

# src/cineclick_churn/api/main.py (part 1: arrencada)
import logging
import os
import time
from contextlib import asynccontextmanager

import mlflow
import pandas as pd
from fastapi import FastAPI, HTTPException

from cineclick_churn.features import construir_features  # LA font única
from cineclick_churn.api.schemas import ClientEntrada, PrediccioSortida

logger = logging.getLogger("cineclick_churn.api")

MODEL_URI = "models:/churn-cineclick@champion"   # el contracte fixat a 03-02
LLINDAR_DEFECTE = float(os.getenv("LLINDAR_ABANDONAMENT", "0.5"))

estat = {}   # aquí viuran el model i les seves metadades durant tota la vida del procés

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- ARRENCADA ---
    # MLFLOW_TRACKING_URI ha de ser a l'entorn; si falta, millor saber-ho JA.
    tracking_uri = os.environ["MLFLOW_TRACKING_URI"]   # KeyError = fallada clara
    mlflow.set_tracking_uri(tracking_uri)

    logger.info("Carregant model des de %s ...", MODEL_URI)
    try:
        estat["model"] = mlflow.sklearn.load_model(MODEL_URI)
        client = mlflow.MlflowClient()
        mv = client.get_model_version_by_alias("churn-cineclick", "champion")
        estat["versio_model"] = mv.version          # "2" a dia d'avui
    except Exception:
        logger.exception("No s'ha pogut carregar el model del registry. Avortant.")
        raise   # fallar ràpid: millor un procés mort que un zombi sense model

    logger.info("Model churn-cineclick v%s carregat.", estat["versio_model"])
    yield   # ---- aquí el servei atén peticions ----
    # --- ATURADA ---
    estat.clear()

app = FastAPI(title="CineClick Churn API", version="0.1.0", lifespan=lifespan)

La nota de resiliència mereix que ens hi aturem: si el registry no respon en arrencar, hi ha dues filosofies. L'equivocada és capturar l'excepció, arrencar "de totes maneres" i retornar errors a cada /predir — un servei que sembla sa però no pot fer la seva feina (un zombi). La correcta, i la que implementem, és fallar ràpid i clar: el raise mata el procés amb un missatge inequívoc al log, i l'orquestrador (Docker a 04-03, Kubernetes a 04-04) ho detectarà i reintentarà o alertarà. Un procés que mor sorollosament es diagnostica en minuts; un zombi silenciós, en dies.

L'endpoint POST /predir

# src/cineclick_churn/api/main.py (part 2: predicció)
@app.post("/predir", response_model=PrediccioSortida)
def predir(client: ClientEntrada, llindar: float | None = None) -> PrediccioSortida:
    """Retorna la probabilitat d'abandonament d'UN client."""
    t0 = time.perf_counter()
    llindar_aplicat = llindar if llindar is not None else LLINDAR_DEFECTE

    # 1. Del JSON validat a un DataFrame d'una fila amb les columnes crues
    df = pd.DataFrame([client.model_dump()])

    # 2. Features: la MATEIXA funció que va fer servir l'entrenament. Ni una línia copiada.
    X = construir_features(df.drop(columns=["id_client"]))

    # 3. Probabilitat de la classe positiva (abandonament)
    prob = float(estat["model"].predict_proba(X)[0, 1])

    # 4. Decisió binària segons el llindar
    sortida = PrediccioSortida(
        id_client=client.id_client,
        probabilitat_abandonament=round(prob, 4),
        prediccio=prob >= llindar_aplicat,
        llindar=llindar_aplicat,
        versio_model=estat["versio_model"],
    )

    latencia_ms = (time.perf_counter() - t0) * 1000
    logger.info("prediccio", extra={"dades_esdeveniment": {
        "id_client": client.id_client,
        "probabilitat": sortida.probabilitat_abandonament,
        "prediccio": sortida.prediccio,
        "versio_model": sortida.versio_model,
        "latencia_ms": round(latencia_ms, 1),
    }})
    return sortida

Pas a pas:

  1. client.model_dump() converteix l'objecte Pydantic (ja validat) en un diccionari, i pd.DataFrame([...]) l'embolcalla en un DataFrame d'una fila — el format que construir_features espera, idèntic al de l'entrenament.
  2. construir_features importada del paquet. Aquest és el moment en què la decisió de 03-03 paga: ratio_tickets = tiquets_suport / antiguitat_mesos.clip(lower=1) es calcula amb la implementació canònica. Si algú la millora demà a features.py, entrenament i serving canvien alhora — el training/serving skew és estructuralment impossible.
  3. predict_proba(X)[0, 1]: fila 0 (només n'hi ha una), columna 1 (probabilitat de la classe positiva). Retornem la probabilitat, no només el booleà, perquè el consumidor pot voler graduar l'oferta.
  4. El llindar és configurable (variable d'entorn LLINDAR_ABANDONAMENT, sobreescrivible per petició amb el query param llindar). El 0.5 per defecte és només el punt de partida matemàtic: el llindar real és una decisió de negoci. Amb el campió actual (precision 0.55, recall 0.68), abaixar el llindar captura més futurs abandonaments a costa d'oferir descomptes a clients que no se n'anaven; apujar-lo fa el contrari. Producte i finances han de triar aquest punt amb un compte de costos — l'API només ha de permetre canviar-lo sense redesplegar.

Healthcheck i versió: GET /salut i GET /versio

# src/cineclick_churn/api/main.py (part 3: operació)
@app.get("/salut")
def salut() -> dict:
    """Healthcheck: el procés és viu I té el model carregat?"""
    if "model" not in estat:
        raise HTTPException(status_code=503, detail="Model no carregat")
    return {"estat": "ok", "model": "churn-cineclick",
            "versio_model": estat["versio_model"]}

@app.get("/versio")
def versio() -> dict:
    """Què corre exactament aquí: versió del servei i del model."""
    return {"servei": app.version,                     # 0.1.0
            "model_uri": MODEL_URI,
            "versio_model": estat.get("versio_model", "desconeguda")}
  • /salut no respon "soc viu" sense més: comprova que el model és a la memòria. Aquest matís importa perquè és l'endpoint que Docker (04-03) i les probes de Kubernetes (04-04) cridaran per decidir si el contenidor rep trànsit o es reinicia. Un healthcheck que menteix és pitjor que no tenir healthcheck.
  • /versio respon a la pregunta operativa per excel·lència: "quin model està servint això ARA?". Quan a 03-02 vam moure l'àlies @champion, vam dir que el rollback era "moure un àlies"; /versio és com verifiques, després de reiniciar el servei, que el canvi ha tingut efecte.

Provar el servei: uvicorn, /docs i curl

Amb el servidor de MLflow corrent (el de sqlite del mòdul 3), arrenquem:

# terminal 1: el servidor MLflow del mòdul 3 (si no està ja corrent)
mlflow server --backend-store-uri sqlite:///mlflow.db --host 127.0.0.1 --port 5000

# terminal 2: el servei de predicció
export MLFLOW_TRACKING_URI=http://127.0.0.1:5000
uvicorn cineclick_churn.api.main:app --host 0.0.0.0 --port 8000

Al log veuràs Carregant model des de models:/churn-cineclick@champion ... i després Model churn-cineclick v2 carregat. — el lifespan en acció. Obre ara http://localhost:8000/docs: FastAPI ha generat la documentació interactiva amb els tres endpoints, els esquemes amb les seves restriccions (antiguitat_mesos >= 0, els valors permesos de pla) i un botó Try it out per llançar peticions des del navegador. Aquest /docs és el contracte que passes a l'equip de producte.

Provem-ho amb curl i un client fictici de perfil arriscat (poca antiguitat, poc ús, tiquets):

curl -s -X POST http://localhost:8000/predir \
  -H "Content-Type: application/json" \
  -d '{
        "id_client": "CLI-04217",
        "antiguitat_mesos": 3,
        "hores_setmana": 1.5,
        "tiquets_suport": 4,
        "pla": "basic",
        "metode_pagament": "targeta",
        "descompte_actiu": false
      }'

Resposta:

{
  "id_client": "CLI-04217",
  "probabilitat_abandonament": 0.8412,
  "prediccio": true,
  "llindar": 0.5,
  "versio_model": "2"
}

I la validació en acció — un pla que no existeix:

curl -s -X POST http://localhost:8000/predir \
  -H "Content-Type: application/json" \
  -d '{"id_client": "CLI-1", "antiguitat_mesos": 3, "hores_setmana": 1.5,
       "tiquets_suport": 4, "pla": "familiar", "metode_pagament": "targeta",
       "descompte_actiu": false}'
# → HTTP 422: "Input should be 'basic', 'estandard' or 'premium'"

El model ni se n'ha assabentat. Això és Pydantic fent de porter.

Una nota sobre seguretat, sense desviar-nos: aquest servei correrà a la xarxa interna de CineClick, però tot i així el mínim exigible és una API key simple (una capçalera X-API-Key verificada amb una dependència de FastAPI contra un secret en variable d'entorn). L'autenticació empresarial seriosa — OAuth2, mTLS, gateway — és terreny de la plataforma de l'empresa, no d'aquesta lliçó; l'important és que l'endpoint no quedi obert a qualsevol ni tan sols "temporalment".

Logging estructurat de cada predicció

Ja ho has vist a l'endpoint: cada predicció emet una línia de log amb camps estructurats, no una frase lliure. Configurem la sortida com a JSON perquè sigui processable per màquines:

# src/cineclick_churn/api/logging_config.py — formatador JSON mínim
import json, logging, hashlib

class FormatadorJSON(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        base = {"ts": self.formatTime(record), "nivell": record.levelname,
                "logger": record.name, "missatge": record.getMessage()}
        base.update(getattr(record, "dades_esdeveniment", {}))
        return json.dumps(base, ensure_ascii=False)

def hash_entrada(payload: dict) -> str:
    """Hash estable de l'input: permet detectar duplicats i auditar sense guardar PII."""
    return hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).hexdigest()[:16]

Cada línia de log de predicció conté: hash de l'input (amb hash_entrada sobre el payload — traçable sense abocar dades del client en text pla), probabilitat, predicció, versió del model i latència en ms. Sembla poc, però és exactament la llavor del monitoratge del mòdul 6: amb aquests logs acumulats podrem pintar la distribució de probabilitats al llarg del temps (deriva, 06-02), la latència p95 (06-01) i el volum de peticions. La regla: si no ho registres des del dia u, no existirà quan ho necessitis.

{"ts": "2026-07-07 11:32:04", "nivell": "INFO", "logger": "cineclick_churn.api",
 "missatge": "prediccio", "id_client": "CLI-04217", "probabilitat": 0.8412,
 "prediccio": true, "versio_model": "2", "latencia_ms": 14.3}

Errors Comuns i Consells

  • Error: carregar el model dins de l'endpoint. Cada petició pagaria segons de càrrega des del registry. El model es carrega una vegada al lifespan i viu a la memòria.
  • Error: reimplementar les features a l'API. "Només és una divisió" — fins que algú canvia el clip a features.py i el serving es queda amb la versió vella: training/serving skew silenciós, el bug més car de detectar. A CineClick està prohibit per decisió de 03-03: sempre from cineclick_churn.features import construir_features.
  • Error: arrencar sense model i retornar errors per petició. El servei zombi. Falla ràpid a l'arrencada (raise al lifespan) i deixa que l'orquestrador faci la seva feina.
  • Error: fer servir async def a l'endpoint de predicció. predict_proba és CPU síncron: dins d'async def bloquejaria el bucle d'esdeveniments i enfonsaria la concurrència. Amb def, FastAPI el porta al threadpool automàticament.
  • Error: acceptar qualsevol string als camps categòrics. Sense Literal, un pla="Bàsic" (amb majúscula i accent) arribaria al model com a categoria desconeguda. L'esquema ha de reflectir el vocabulari exacte de l'entrenament.
  • Consell: retorna sempre la versió del model a la resposta. Costa un camp i converteix cada resposta en evidència auditable.
  • Consell: tracta el llindar com a configuració de negoci, no com a constant del codi. Canviar de 0.5 a 0.4 no hauria de requerir un desplegament.

Exercicis

Exercici 1

Producte demana un endpoint POST /predir-lot que rebi una llista de clients (màxim 100) i retorni una llista de prediccions, per puntuar d'una vegada tots els assistents a un webinar de retenció. Implementa'l reutilitzant els esquemes existents. Pista: predict_proba accepta un DataFrame de N files.

Exercici 2

Escriu un test amb TestClient de FastAPI (a tests/test_api.py) que verifiqui que una petició amb antiguitat_mesos: -5 rep un 422 sense arribar a invocar el model. Pista: pots fer monkeypatch d'estat per no dependre del registry als tests.

Exercici 3

Ara mateix, si MLflow està caigut quan arrenca el servei, el procés mor (correcte). Però què passa si MLflow cau després, amb el servei ja arrencat? Raona: (a) continua funcionant /predir?, (b) què retorna /salut?, (c) és acceptable aquest comportament?

Solucions

Solució 1

# a schemas.py
class LotEntrada(BaseModel):
    clients: list[ClientEntrada] = Field(..., min_length=1, max_length=100)

# a main.py
@app.post("/predir-lot", response_model=list[PrediccioSortida])
def predir_lot(lot: LotEntrada) -> list[PrediccioSortida]:
    df = pd.DataFrame([c.model_dump() for c in lot.clients])
    X = construir_features(df.drop(columns=["id_client"]))
    probs = estat["model"].predict_proba(X)[:, 1]   # UN sol predict per a N files
    return [
        PrediccioSortida(
            id_client=c.id_client,
            probabilitat_abandonament=round(float(p), 4),
            prediccio=float(p) >= LLINDAR_DEFECTE,
            llindar=LLINDAR_DEFECTE,
            versio_model=estat["versio_model"],
        )
        for c, p in zip(lot.clients, probs)
    ]

L'essencial: una sola crida a construir_features i a predict_proba per a les N files, no un bucle de N prediccions — la diferència de rendiment la quantificarem a 04-05. El max_length=100 protegeix el servei que algú li enviï la base sencera per HTTP (per a això hi ha el batch de 04-01).

Solució 2

# tests/test_api.py
from fastapi.testclient import TestClient
from cineclick_churn.api import main

def test_antiguitat_negativa_retorna_422(monkeypatch):
    # Model fals a l'estat: si l'endpoint el cridés, el test explotaria
    class ModelProhibit:
        def predict_proba(self, X):
            raise AssertionError("El model no s'ha d'invocar amb entrada invàlida")

    monkeypatch.setitem(main.estat, "model", ModelProhibit())
    monkeypatch.setitem(main.estat, "versio_model", "test")

    client_http = TestClient(main.app)   # sense executar el lifespan real
    resposta = client_http.post("/predir", json={
        "id_client": "CLI-X", "antiguitat_mesos": -5, "hores_setmana": 2,
        "tiquets_suport": 0, "pla": "basic", "metode_pagament": "targeta",
        "descompte_actiu": False,
    })
    assert resposta.status_code == 422

Si Pydantic deixés passar l'entrada, ModelProhibit.predict_proba llançaria l'AssertionError i el test fallaria amb un missatge clar. Aquest test entrarà al CI de la lliçó 05-01.

Solució 3

(a) Sí. El model ja és a la memòria; /predir no toca MLflow en el camí de la petició. Que el registry estigui caigut no afecta el trànsit en curs — és una virtut del patró "carregar en arrencar". (b) /salut retorna 200 ok, perquè només comprova que el model és a la memòria, no la connectivitat amb MLflow — i això és correcte: el healthcheck ha de mesurar la capacitat de fer la feina (predir), no la salut de dependències que ja no necessita. (c) Acceptable amb un matís: les noves arrencades (un reinici, una rèplica nova en escalar) sí que fallaran mentre MLflow estigui caigut. És un risc conegut que apuntem per a 04-03: allà discutirem l'alternativa d'enfornar el model dins la imatge justament per eliminar aquesta dependència a l'arrencada, i per què CineClick tot i així tria el registry.

Conclusió

El model de la Laura ja atén peticions. El servei té contracte explícit (Pydantic valida domini, no només tipus), carrega el campió una única vegada des de models:/churn-cineclick@champion amb política de fallar ràpid, prediu important construir_features de la font única — zero lògica duplicada —, exposa /salut i /versio perquè les màquines i els humans sàpiguen què corre, i registra cada predicció en logs JSON que el mòdul 6 convertirà en monitoratge. Tot això funciona... a la teva màquina, amb el teu Python 3.11.9, el teu requirements.lock instal·lat i la teva variable d'entorn apuntant al teu MLflow. El servidor de producció no té res de tot això, i "instal·la-ho a mà i creua els dits" és exactament la mena de procés que aquest curs va venir a eliminar. La lliçó següent congela el servei complet — sistema operatiu, Python, dependències i codi — en una imatge Docker que corre idèntica a qualsevol lloc: el "funciona a la meva màquina" és a punt de deixar de ser una excusa.

◄ Anterior
◄
Següent ►
►
El Projecte | Sobre nosaltres | Contribuir | Donacions | Llicència
© Copyright 2026. Tots els drets reservats