En ingénierie de Machine Learning et MLOps, calculer une prédiction au vol ne suffit pas. Si un client conteste une décision automatisée d'octroi de crédit, ou si le modèle commence à commettre des erreurs massives après six mois en production, vous devez impérativement être en mesure de répondre à cette question :
« Quelles étaient exactement les variables d'entrée soumises au modèle le 14 mai à 14h32, et quelle probabilité a été calculée ? »
Pour assurer cette traçabilité (Audit Trail) et collecter les données nécessaires à la détection de la dérive des données (Data Drift), nous allons intégrer une base de données relationnelle à notre API Flask grâce à Flask-SQLAlchemy.
🛠️ Pourquoi SQLAlchemy en Contexte Data ?#
SQLAlchemy est l'ORM (Object-Relational Mapping) de référence en Python. Il vous permet de manipuler les tables et enregistrements d'une base de données comme de simples classes et objets Python, sans avoir à écrire de requêtes SQL textuelles brutes pour les opérations courantes.
De plus, l'extension Flask-SQLAlchemy gère automatiquement les pools de connexions, les sessions transactionnelles et permet de basculer de SQLite (idéal pour le développement local rapide) vers PostgreSQL (le standard de l'entreprise en production) en modifiant une seule ligne de configuration !
1. Installer l'Extension Flask-SQLAlchemy#
Dans votre environnement virtuel actif, installez le package :
2. Définir le Modèle de Données d'Audit (app/models/prediction_log.py)#
Nous concevons une table dédiée, prediction_logs, capable d'enregistrer chaque appel à notre API :
from datetime import datetime, timezone
import json
from app import db
class PredictionLog(db.Model):
"""Table de traçabilité et d'audit pour chaque prédiction d'inférence."""
__tablename__ = "prediction_logs"
id = db.Column(db.Integer, primary_key=True)
client_id = db.Column(db.String(64), index=True, nullable=False)
# Stockage de l'ensemble des variables d'entrée sous forme de JSON textuel
features_json = db.Column(db.Text, nullable=False)
# Résultat de l'inférence
prediction = db.Column(db.Integer, nullable=False)
probability = db.Column(db.Float, nullable=False)
# Horodatage UTC précis
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), nullable=False)
def to_dict(self):
"""Convertit l'enregistrement en dictionnaire sérialisable en JSON."""
return {
"id": self.id,
"client_id": self.client_id,
"features": json.loads(self.features_json),
"prediction": self.prediction,
"probability": round(self.probability, 4),
"created_at": self.created_at.isoformat()
}
3. Initialiser la Base de Données dans l'Application#
Nous initialisons l'objet db dans notre Application Factory :
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
def create_app(config_name="development"):
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///scoring_audit.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
# Liaison de l'ORM à l'application Flask
db.init_app(app)
# Enregistrement des routes
from app.api.routes import api_bp
app.register_blueprint(api_bp, url_prefix="/api/v1")
# Création des tables si elles n'existent pas encore
with app.app_context():
db.create_all()
return app
4. Persister Automatiquement le Score lors de l'Appel /predict#
Mettons à jour notre endpoint d'inférence pour qu'il sauvegarde chaque transaction en base avant de renvoyer la réponse :
import json
from flask import Blueprint, jsonify, request
from app import db
from app.models.prediction_log import PredictionLog
api_bp = Blueprint("api", __name__)
@api_bp.route("/predict", methods=["POST"])
def predict():
payload = request.get_json()
if not payload or "client_id" not in payload:
return jsonify({"erreur": "Clé 'client_id' manquante dans la requête."}), 400
client_id = payload["client_id"]
# Simulation du calcul prédictif par le modèle
probabilite = 0.78
classe_predite = 1 if probabilite >= 0.50 else 0
# 1. Création de l'enregistrement d'audit
log_entry = PredictionLog(
client_id=str(client_id),
features_json=json.dumps(payload),
prediction=classe_predite,
probability=probabilite
)
# 2. Sauvegarde dans la base de données
db.session.add(log_entry)
db.session.commit()
# 3. Réponse renvoyée au client
return jsonify({
"status": "success",
"log_id": log_entry.id,
"client_id": client_id,
"prediction": classe_predite,
"probability": probabilite
}), 200
5. Exposer un Endpoint de Consultation des Logs d'Audit#
Pour permettre à votre équipe de Data Science de vérifier les dernières prédictions générées par l'API sans avoir à ouvrir un client SQL externe, créons une route de monitoring :
@api_bp.route("/audit/recent", methods=["GET"])
def get_recent_logs():
"""Renvoie les 10 dernières prédictions effectuées par le modèle."""
# Requête ORM triée par date décroissante
recent_logs = PredictionLog.query.order_by(PredictionLog.created_at.desc()).limit(10).all()
return jsonify({
"total_extraits": len(recent_logs),
"logs": [log.to_dict() for log in recent_logs]
}), 200
En effectuant une requête GET sur http://localhost:5000/api/v1/audit/recent, vous obtenez instantanément le journal d'activité de votre modèle au format JSON !
Votre API est désormais capable d'effectuer des calculs analytiques et de mémoriser rigoureusement l'ensemble de ses décisions. Dans la deuxième partie du cours, nous apprendrons à créer des interfaces graphiques de restitution, à charger un modèle réel entraîné avec Scikit-learn et à conteneuriser le tout dans Docker.
Commentaires