Workflows data avec Code Interpreter
Workflows data avec Code Interpreter
Code Interpreter prend toute sa dimension dans les workflows complets : import de donnees, nettoyage, analyse, visualisation et export. Cette lecon assemble les pieces vues precedemment en pipelines de bout en bout.
Workflow 1 : rapport financier automatise
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": (
"Genere un rapport financier complet a partir de ces 3 fichiers : "
"1. SYNTHESE : CA total, marge brute, resultat net "
"2. ANALYSE MENSUELLE : evolution mois par mois avec variations "
"3. ECARTS BUDGET : comparaison reel vs previsionnel, ecarts significatifs "
"4. GRAPHIQUES : "
" - Evolution CA et depenses (double axe) "
" - Waterfall des ecarts budgetaires "
" - Repartition des depenses par categorie "
"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.3",
input=[{"role": "user", "content": content}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Produis un rapport professionnel. "
"Affiche les chiffres avec separateurs de milliers. "
"Utilise le format europeen (virgule decimale). "
"Commente chaque ecart significatif (> 5%)."
)
)
# Recuperer tous les fichiers generes
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"Genere : {f.name}")Workflow 2 : analyse de cohortes
with open("utilisateurs.csv", "rb") as f:
fichier = client.files.create(file=f, purpose="assistants")
response = client.responses.create(
model="gpt-5.3",
input=[{
"role": "user",
"content": [
{"type": "text", "text": (
"Effectue une analyse de cohortes sur ces donnees utilisateurs : "
"1. Definis 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 estimee 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 etape par etape. "
"Affiche les resultats intermediaires. "
"Pour la heatmap, utilise une palette verte (sombre = bon, clair = faible). "
"Annote chaque cellule avec le pourcentage."
)
)Workflow 3 : scoring et segmentation
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.3",
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, frequence "
"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 : previsions et tendances
response = client.responses.create(
model="gpt-5.3",
input=[{
"role": "user",
"content": [
{"type": "text", "text": (
"A partir de ces donnees de ventes mensuelles : "
"1. Decompose la serie temporelle (tendance, saisonnalite, residu) "
"2. Applique un lissage exponentiel (Holt-Winters) "
"3. Prevois les 6 prochains mois avec intervalle de confiance 95% "
"4. Compare avec une regression lineaire simple "
"5. Genere un graphique avec historique + previsions "
"6. Calcule les metriques d'erreur (MAE, RMSE, MAPE)"
)},
{"type": "input_file", "file_id": fichier.id}
]
}],
tools=[{"type": "code_interpreter"}],
instructions=(
"Utilise statsmodels pour la decomposition et les previsions. "
"Colorie l'intervalle de confiance en bleu transparent. "
"Marque clairement la separation historique/previsions."
)
)Orchestrer un workflow multi-etapes
Pour les workflows complexes, decoupez en etapes avec passage de fichiers :
class WorkflowData:
def __init__(self):
self.client = OpenAI()
self.fichiers_intermediaires = []
def etape(self, instruction: str, fichiers_ids: list[str] = None) -> str:
"""Execute une etape 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.3",
input=[{"role": "user", "content": content}],
tools=[{"type": "code_interpreter"}]
)
# Capturer les fichiers generes
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()
# Etape 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, gere les valeurs manquantes, "
"normalise les types. Exporte le resultat.",
[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 cles a retenir
- Decoupez les workflows complexes en etapes avec passage de fichiers
- Utilisez des instructions detaillees pour chaque etape
- Le modele gere nativement les analyses statistiques et le machine learning basique
- Recuperez systematiquement les fichiers intermediaires pour la tracabilite
- Combinez analyse de donnees et visualisation dans un meme appel pour des rapports complets