Versioning et rollback
Pourquoi versionner vos modèles
En production, vous allez itérer sur votre modèle fine-tuné : nouvelles données, hyperparamètres ajustés, nouveau modèle de base. Sans un système de versioning rigoureux, vous perdrez la traçabilité et la capacité de revenir en arrière en cas de régression.
Convention de nommage
Suffixes versionnés
Utilisez le paramètre suffix pour identifier chaque version :
# Convention : {nom}-v{version_majeure}.{version_mineure}
job = client.fine_tuning.jobs.create(
training_file=training_file.id,
model="gpt-5.3-mini",
suffix="support-client-v2.1"
)Schéma de version recommandé
- v1.0 : premier modèle validé en production
- v1.1 : correction de données (mêmes hyperparamètres)
- v2.0 : changement de modèle de base ou de stratégie
- v2.1 : ajustement d’hyperparamètres
Registre de modèles
Maintenez un registre centralisé de tous vos modèles :
import json
import datetime
class ModelRegistry:
"""Registre des modèles fine-tunés."""
def __init__(self, chemin: str = "model_registry.json"):
self.chemin = chemin
try:
with open(chemin, "r") as f:
self.registre = json.load(f)
except FileNotFoundError:
self.registre = {"modeles": [], "production": None}
def enregistrer(
self,
version: str,
model_id: str,
job_id: str,
dataset_info: dict,
hyperparams: dict,
metriques: dict,
notes: str = ""
):
"""Enregistre un nouveau modèle dans le registre."""
entree = {
"version": version,
"model_id": model_id,
"job_id": job_id,
"date": datetime.datetime.now().isoformat(),
"dataset": dataset_info,
"hyperparametres": hyperparams,
"metriques": metriques,
"notes": notes,
"statut": "candidat"
}
self.registre["modeles"].append(entree)
self._sauvegarder()
print(f"Modèle {version} enregistré : {model_id}")
def promouvoir(self, version: str):
"""Promeut un modèle en production."""
ancien = self.registre["production"]
for m in self.registre["modeles"]:
if m["version"] == version:
m["statut"] = "production"
self.registre["production"] = version
break
if ancien:
for m in self.registre["modeles"]:
if m["version"] == ancien:
m["statut"] = "retiré"
self._sauvegarder()
print(f"Production : {ancien} -> {version}")
def rollback(self, version: str):
"""Rollback vers une version précédente."""
self.promouvoir(version)
print(f"Rollback effectué vers {version}")
def modele_production(self) -> str:
"""Retourne l'ID du modèle en production."""
version = self.registre["production"]
for m in self.registre["modeles"]:
if m["version"] == version:
return m["model_id"]
return None
def historique(self):
"""Affiche l'historique des modèles."""
print(f"{'Version':<12} {'Statut':<12} {'Date':<12} {'Loss val':<10}")
print("-" * 50)
for m in self.registre["modeles"]:
date = m["date"][:10]
loss = m["metriques"].get("validation_loss", "N/A")
print(f"{m['version']:<12} {m['statut']:<12} {date:<12} {loss}")
def _sauvegarder(self):
with open(self.chemin, "w") as f:
json.dump(self.registre, f, indent=2, ensure_ascii=False)Utilisation du registre
registry = ModelRegistry()
# Enregistrer un nouveau modèle
registry.enregistrer(
version="v2.1",
model_id="ft:gpt-5.3-mini:org::support-v2.1:abc123",
job_id="ftjob-xyz789",
dataset_info={"train": 350, "validation": 45},
hyperparams={"n_epochs": 3, "lr": "auto"},
metriques={"validation_loss": 0.38, "precision_test": 0.93},
notes="Ajout de 50 exemples correctifs pour les remboursements"
)
# Promouvoir en production
registry.promouvoir("v2.1")
# En cas de problème, rollback
registry.rollback("v2.0")Rollback en production
Mécanisme de rollback rapide
import os
def deployer_modele(version: str, registry: ModelRegistry):
"""Déploie un modèle en mettant à jour la variable d'environnement."""
model_id = None
for m in registry.registre["modeles"]:
if m["version"] == version:
model_id = m["model_id"]
break
if not model_id:
raise ValueError(f"Version {version} non trouvée")
# Mettre à jour la configuration
os.environ["OPENAI_FT_MODEL"] = model_id
registry.promouvoir(version)
print(f"Modèle déployé : {version} ({model_id})")
return model_id
def rollback_rapide(registry: ModelRegistry):
"""Rollback vers la dernière version stable."""
production = registry.registre["production"]
versions = [
m for m in registry.registre["modeles"]
if m["version"] != production and m["statut"] != "retiré"
]
if not versions:
# Fallback sur le modèle de base
os.environ["OPENAI_FT_MODEL"] = "gpt-5.3-mini"
print("Rollback vers le modèle de base")
return
derniere_stable = versions[-1]
deployer_modele(derniere_stable["version"], registry)Sauvegarder les données d’entraînement
Versionnez aussi vos données, pas seulement vos modèles :
import shutil
def archiver_dataset(version: str, fichiers: list[str], dossier_archive: str):
"""Archive les données d'entraînement pour une version donnée."""
dossier = f"{dossier_archive}/{version}"
os.makedirs(dossier, exist_ok=True)
for fichier in fichiers:
shutil.copy2(fichier, dossier)
print(f"Données archivées dans {dossier}")
archiver_dataset(
"v2.1",
["train.jsonl", "validation.jsonl", "test.jsonl"],
"archives/datasets"
)Automatiser le pipeline
Script de déploiement complet
def pipeline_deploiement(
training_file: str,
validation_file: str,
version: str,
modele_base: str = "gpt-5.3-mini",
hyperparams: dict = None
):
"""Pipeline complet : upload, train, évaluer, enregistrer."""
client = OpenAI()
registry = ModelRegistry()
# 1. Upload
print("Upload des données...")
train = client.files.create(file=open(training_file, "rb"), purpose="fine-tune")
val = client.files.create(file=open(validation_file, "rb"), purpose="fine-tune")
# 2. Entraînement
print("Lancement du fine-tuning...")
params = {"training_file": train.id, "validation_file": val.id,
"model": modele_base, "suffix": version}
if hyperparams:
params["hyperparameters"] = hyperparams
job = client.fine_tuning.jobs.create(**params)
# 3. Attendre la fin
import time
while True:
job = client.fine_tuning.jobs.retrieve(job.id)
if job.status in ("succeeded", "failed"):
break
time.sleep(30)
if job.status == "failed":
print(f"Échec : {job.error}")
return None
# 4. Enregistrer
registry.enregistrer(
version=version,
model_id=job.fine_tuned_model,
job_id=job.id,
dataset_info={"train": training_file, "validation": validation_file},
hyperparams=hyperparams or {},
metriques={"trained_tokens": job.trained_tokens}
)
# 5. Archiver les données
archiver_dataset(version, [training_file, validation_file], "archives/datasets")
print(f"Pipeline terminé : {job.fine_tuned_model}")
return job.fine_tuned_modelSupprimer les anciens modèles
Les modèles fine-tunés sont stockés chez OpenAI et peuvent être supprimés quand vous n’en avez plus besoin :
def nettoyer_anciens_modeles(registry: ModelRegistry, garder: int = 3):
"""Supprime les anciens modèles en gardant les N derniers."""
modeles = registry.registre["modeles"]
production = registry.registre["production"]
candidats = [
m for m in modeles
if m["version"] != production
]
a_supprimer = candidats[:-garder] if len(candidats) > garder else []
for m in a_supprimer:
try:
client.models.delete(m["model_id"])
print(f"Supprimé : {m['version']} ({m['model_id']})")
except Exception as e:
print(f"Erreur suppression {m['version']} : {e}")Points clés à retenir
- Utilisez des suffixes versionnés pour identifier chaque modèle
- Maintenez un registre centralisé avec métriques et métadonnées
- Implémentez un mécanisme de rollback rapide
- Archivez les données d’entraînement avec chaque version
- Nettoyez régulièrement les anciens modèles pour éviter les coûts inutiles
- Automatisez le pipeline pour réduire les erreurs humaines