Workflows data avec Code Interpreter
Workflows data avec Code Interpreter
Les leçons précédentes ont traité les gestes un par un. Un vrai travail d’analyse les enchaîne : on importe, on nettoie, on calcule, on visualise, on exporte — et c’est l’enchaînement, pas chaque geste isolé, qui prend du temps quand il est fait à la main.
Les quatre workflows qui suivent sont des cas réels et volontairement différents : un reporting périodique, une analyse de comportement dans le temps, une segmentation, une projection. Lisez-les moins pour le code que pour la forme des prompts — numérotés, explicites sur les livrables attendus, précis sur le format. C’est cette forme qui transforme une demande vague en rapport exploitable, et elle se transpose telle quelle à vos propres données.
Workflow 1 : rapport financier automatisé
Le reporting périodique est le meilleur candidat à l’automatisation, pour une raison simple : les fichiers changent, la structure du rapport non. Une fois le prompt écrit et validé sur un mois, il vaut pour les onze suivants. Notez la dernière consigne, qui demande un export Excel par section — c’est ce qui rend le résultat directement transmissible, plutôt qu’un texte qu’il faudrait remettre en forme.
from openai import OpenAI
client = OpenAI()
# Upload des fichiers sources
fichiers = {}
for nom in ["revenus_2025.xlsx", "depenses_2025.xlsx", "budget_previsionnel.xlsx"]:
with open(nom, "rb") as f:
fichier = client.files.create(file=f, purpose="assistants")
fichiers[nom] = fichier.id
content = [
{"type": "text", "text": (
"Génère un rapport financier complet a partir de ces 3 fichiers : "
"1. SYNTHESE : CA total, marge brute, résultat net "
"2. ANALYSE MENSUELLE : évolution mois par mois avec variations "
"3. ECARTS BUDGET : comparaison reel vs previsionnel, ecarts significatifs "
"4. GRAPHIQUES : "
" - Évolution CA et dépenses (double axe) "
" - Waterfall des écarts budgétaires "
" - Répartition des dépenses par catégorie "
"5. EXPORT : fichier Excel avec un onglet par section"
)}
]
for fid in fichiers.values():
content.append({"type": "input_file", "file_id": fid})
response = client.responses.create(
model="gpt-5.6-terra",
input=[{"role": "user", "content": content}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Produis un rapport professionnel. "
"Affiche les chiffres avec séparateurs de milliers. "
"Utilise le format européen (virgule décimale). "
"Commente chaque ecart significatif (> 5%)."
)
)
# Récupérer tous les fichiers générés
for item in response.output:
if item.type == "code_interpreter_call":
for result in item.results:
if hasattr(result, "files"):
for f in result.files:
contenu = client.files.content(f.file_id)
with open(f"sortie_{f.name}", "wb") as out:
out.write(contenu.read())
print(f"Génère : {f.name}")Workflow 2 : analyse de cohortes
L’analyse de cohortes suit dans le temps des groupes constitués à une même date d’entrée. C’est un calcul que peu d’équipes font régulièrement parce qu’il est fastidieux à monter, et c’est précisément pour cela qu’il gagne à être délégué : la logique est standard, seules les données changent.
with open("utilisateurs.csv", "rb") as f:
fichier = client.files.create(file=f, purpose="assistants")
response = client.responses.create(
model="gpt-5.6-terra",
input=[{
"role": "user",
"content": [
{"type": "text", "text": (
"Effectue une analyse de cohortes sur ces données utilisateurs : "
"1. Définis les cohortes par mois d'inscription "
"2. Calcule le taux de retention par cohorte (mois 1, 2, ..., 12) "
"3. Cree une heatmap de retention "
"4. Identifie la cohorte avec la meilleure retention "
"5. Calcule la LTV estimée par cohorte "
"6. Exporte le tableau de retention en CSV"
)},
{"type": "input_file", "file_id": fichier.id}
]
}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Montre le code étape par étape. "
"Affiche les résultats intermédiaires. "
"Pour la heatmap, utilise une palette verte (sombre = bon, clair = faible). "
"Annote chaque cellule avec le pourcentage."
)
)Workflow 3 : scoring et segmentation
Ici le modèle ne se contente plus de décrire les données : il construit un score et forme des groupes. Une précaution s’impose donc — demandez systématiquement la méthode en plus du résultat. Un segment « clients à risque » sans les critères qui l’ont produit est inexploitable : personne ne peut le contester, le reproduire, ni le corriger.
def segmenter_clients(fichier_path: str) -> dict:
"""Segmente les clients avec RFM et clustering."""
with open(fichier_path, "rb") as f:
fichier = client.files.create(file=f, purpose="assistants")
response = client.responses.create(
model="gpt-5.6-terra",
input=[{
"role": "user",
"content": [
{"type": "text", "text": (
"Realise une segmentation client RFM : "
"1. Calcule Recency, Frequency, Monetary pour chaque client "
"2. Attribue un score RFM de 1 a 5 pour chaque dimension "
"3. Cree des segments : Champions, Loyaux, A risque, Perdus "
"4. Applique K-Means (k=4 a 8, choisis le meilleur k avec elbow) "
"5. Visualise les segments en 2D (PCA ou t-SNE) "
"6. Profil de chaque segment : taille, CA moyen, fréquence "
"7. Exporte : CSV avec segment par client + rapport PDF-like"
)},
{"type": "input_file", "file_id": fichier.id}
]
}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Utilise sklearn pour le clustering. "
"Normalise les variables avant le clustering. "
"Affiche le graphique elbow pour justifier le choix de k. "
"Cree un radar chart par segment."
)
)
fichiers_sortie = []
for item in response.output:
if item.type == "code_interpreter_call":
for result in item.results:
if hasattr(result, "files"):
for f in result.files:
contenu = client.files.content(f.file_id)
with open(f.name, "wb") as out:
out.write(contenu.read())
fichiers_sortie.append(f.name)
return {
"analyse": response.output_text,
"fichiers": fichiers_sortie
}Workflow 4 : prévisions et tendances
La prévision est le workflow où la prudence compte le plus. Le container sait faire tourner une régression ou une décomposition saisonnière, et il le fera même quand vos données ne s’y prêtent pas — vingt points de mesure suffisent à produire une courbe parfaitement présentable et parfaitement dénuée de sens. Exigez l’intervalle de confiance et les hypothèses du modèle : c’est ce qui distingue une projection d’une extrapolation.
response = client.responses.create(
model="gpt-5.6-terra",
input=[{
"role": "user",
"content": [
{"type": "text", "text": (
"A partir de ces données de ventes mensuelles : "
"1. Decompose la série temporelle (tendance, saisonnalité, residu) "
"2. Applique un lissage exponentiel (Holt-Winters) "
"3. Prevois les 6 prochains mois avec intervalle de confiance 95% "
"4. Compare avec une régression linéaire simple "
"5. Génère un graphique avec historique + previsions "
"6. Calcule les métriques d'erreur (MAE, RMSE, MAPE)"
)},
{"type": "input_file", "file_id": fichier.id}
]
}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Utilise statsmodels pour la décomposition et les prévisions. "
"Colorie l'intervalle de confiance en bleu transparent. "
"Marque clairement la séparation historique/prévisions."
)
)Orchestrer un workflow multi-étapes
Au-delà d’une certaine taille, un prompt unique atteint ses limites : l’analyse dérive, une consigne intermédiaire est oubliée, et vous ne savez pas à quelle étape. Le découpage en appels successifs, chacun recevant les fichiers produits par le précédent, restaure ce que vous aviez perdu — un point de contrôle après chaque étape, et la possibilité de reprendre à l’étape fautive au lieu de tout relancer.
class WorkflowData:
def __init__(self):
self.client = OpenAI()
self.fichiers_intermediaires = []
def etape(self, instruction: str, fichiers_ids: list[str] = None) -> str:
"""Exécute une étape du workflow."""
if fichiers_ids is None:
fichiers_ids = [f for f in self.fichiers_intermediaires[-3:]]
content = [{"type": "text", "text": instruction}]
for fid in fichiers_ids:
content.append({"type": "input_file", "file_id": fid})
response = self.client.responses.create(
model="gpt-5.6-terra",
input=[{"role": "user", "content": content}],
tools=[{"type": "code_interpreter"}]
)
# Capturer les fichiers générés
for item in response.output:
if item.type == "code_interpreter_call":
for result in item.results:
if hasattr(result, "files"):
for f in result.files:
self.fichiers_intermediaires.append(f.file_id)
return response.output_text
def telecharger_resultats(self, dossier_sortie: str):
"""Telecharge tous les fichiers generes."""
import os
os.makedirs(dossier_sortie, exist_ok=True)
for fid in self.fichiers_intermediaires:
try:
contenu = self.client.files.content(fid)
chemin = os.path.join(dossier_sortie, f"{fid}.bin")
with open(chemin, "wb") as out:
out.write(contenu.read())
except Exception:
pass
# Utilisation
wf = WorkflowData()
# Étape 1 : import et nettoyage
with open("donnees_brutes.csv", "rb") as f:
fichier = client.files.create(file=f, purpose="assistants")
wf.etape(
"Nettoie ce CSV : supprime doublons, gère les valeurs manquantes, "
"normalise les types. Exporte le résultat.",
[fichier.id]
)
# Etape 2 : analyse
wf.etape("Calcule les KPIs principaux et identifie les tendances.")
# Etape 3 : visualisation
wf.etape("Cree 4 graphiques cles pour un rapport de direction.")
# Etape 4 : export
wf.etape("Compile tout dans un fichier Excel avec onglets et graphiques integres.")
wf.telecharger_resultats("./sortie_workflow/")Points clés à retenir
- Découpez les workflows complexes en étapes avec passage de fichiers
- Utilisez des instructions détaillées pour chaque étape
- Le modèle gère nativement les analyses statistiques et le machine learning basique
- Récupérez systématiquement les fichiers intermédiaires pour la traçabilité
- Combinez analyse de données et visualisation dans un même appel pour des rapports complets