Navigation : Lab 13 << | Index | >> Lab 15
À la fin de ce laboratoire, vous saurez : 1. Identifier les blocs de code dans un pipeline ML 2. Réaliser des études d’ablation pour prioriser les améliorations 3. Raffiner de manière ciblée les composants critiques 4. Optimiser l’effort d’amélioration d’un modèle
Pourquoi ce lab s’appuie sur un provider LLM abstrait (LiteLLM/OpenRouter) plutôt qu’un API figée : le pipeline d’ablation interroge un LLM à chaque étape (identification des blocs, estimation d’importance, raffinement). Coder un appel openai.ChatCompletion en dur verrouillerait le notebook sur un seul fournisseur — ici l’abstraction provider: openrouter permet de changer de modèle (GPT, Claude, Qwen) en éditant un paramètre. C’est la robustesse attendue d’un lab reproduit sur des environnements hétérogènes : la logique d’ablation est découplée du moteur LLM sous-jacent.
import sys
sys.path.insert(0, '..')
import json
import re
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
from config import get_settings
from utils import LLMClient
print("Imports OK : json, re, numpy, dataclasses, Enum, config, utils")Imports OK : json, re, numpy, dataclasses, Enum, config, utilsChargement des paramètres de configuration.
Pourquoi décomposer un pipeline ML en blocs logiques avant d’ablater : l’ablation consiste à mesurer l’impact de chaque composant en le supprimant. Mais un script ML n’est pas une liste de fonctions isolées — c’est un flux continu (imports → chargement → feature engineering → entraînement → évaluation). Le CodeBlockAnalyzer découpe ce flux en blocs sémantiques (Data Loading, Feature Engineering, Training…) pour que l’ablation opère à la bonne granularité : ni trop fine (ablater une ligne) ni trop grossière (ablater la moitié du pipeline). Sans cette décomposition, mesurer « l’importance d’un bloc » n’aurait pas d’unité de décision.
@dataclass
class CodeBlock:
id: str
name: str
description: str
code: str
importance: float = 0.0
class BlockType(Enum):
FEATURE_ENGINEERING = "feature_engineering"
MODEL_SELECTION = "model_selection"
HYPERPARAMETER_TUNING = "hyperparameter_tuning"
TRAINING = "training"
EVALUATION = "evaluation"
ENSEMBLE = "ensemble"
print("Dataclasses definies : CodeBlock (id, nom, description, code, importance), BlockType (6 types de blocs ML)")Dataclasses definies : CodeBlock (id, nom, description, code, importance), BlockType (6 types de blocs ML)Analyseur de blocs de code pour identifier les composants a tester.
class CodeBlockAnalyzer:
"""Identifie les blocs de code dans un pipeline ML."""
def __init__(self, llm: LLMClient):
self.llm = llm
def identify_blocks(self, code: str) -> List[CodeBlock]:
"""Identifie les blocs logiques dans le code."""
prompt = f"""Analyse ce code ML et identifie les blocs logiques.
CODE:
```python
{code[:2000]}
```
Identifie 3-5 blocs (feature engineering, model, training, etc.).
Format JSON:
[
{{"id": "block1", "name": "...", "description": "...", "code": "..."}}]
]
JSON:"""
response = self.llm.generate(prompt, temperature=0.2)
try:
match = re.search(r'\[.*\]', response, re.DOTALL)
if match:
data = json.loads(match.group(0))
return [CodeBlock(**b) for b in data]
except:
pass
return []
print("Classe CodeBlockAnalyzer definie : identification de blocs logiques dans un pipeline ML via LLM")Classe CodeBlockAnalyzer definie : identification de blocs logiques dans un pipeline ML via LLMPourquoi estimer l’importance sur une échelle 0.0–1.0 plutôt qu’exécuter réellement l’ablation : une ablation ML rigoureuse ré-entraînerait le modèle sans chaque bloc et comparerait les métriques — coûteux (N ré-entraînements) et irréalisable dans un lab pédagogique. Le simulateur demande au LLM d’estimer l’importance relative de chaque bloc à partir de son rôle sémantique : « Data Loading & Feature Engineering » est vraisemblablement plus critique que « Imports ». C’est une approximation experte qui produit un classement actionnable (quel bloc raffiner en priorité) sans le coût du benchmark complet. La rigueur scientifique exigerait de valider ensuite ce classement par une vraie ablation — le lab enseigne la méthode de priorisation, pas le verdict définitif.
class AblationStudy:
"""Simule des etudes d'ablation pour prioriser les ameliorations."""
def __init__(self, llm: LLMClient):
self.llm = llm
def estimate_importance(self, blocks: List[CodeBlock], task: str) -> List[CodeBlock]:
"""Estime l'importance relative de chaque bloc."""
blocks_desc = "\n".join([
f"- {b.id}: {b.name} - {b.description[:50]}"
for b in blocks
])
prompt = f"""Estime l'importance de chaque bloc pour la tache: {task}
BLOCS:
{blocks_desc}
Donne un score d'importance de 0.0 a 1.0 pour chaque bloc.
Format JSON:
[
{{"id": "block1", "importance": 0.9}},
{{"id": "block2", "importance": 0.6}}
]
JSON:"""
response = self.llm.generate(prompt, temperature=0.2)
try:
match = re.search(r'\[.*\]', response, re.DOTALL)
if match:
data = json.loads(match.group(0))
importance_map = {d['id']: d['importance'] for d in data}
for block in blocks:
block.importance = importance_map.get(block.id, 0.5)
except:
# Default importance
for i, block in enumerate(blocks):
block.importance = 0.8 - i * 0.1
return sorted(blocks, key=lambda b: b.importance, reverse=True)
print("Classe AblationStudy definie : estimation de l'importance relative des blocs (0.0-1.0) via LLM")Classe AblationStudy definie : estimation de l'importance relative des blocs (0.0-1.0) via LLMPourquoi raffiner uniquement le bloc prioritaire plutôt que tout le pipeline : l’ablation a identifié qu’un seul bloc concentre l’essentiel de la valeur (le bloc d’importance maximale). Raffiner ce bloc précis — et lui seul — maximise le retour sur effort : le TargetedRefiner demande au LLM de réécrire ce bloc avec des améliorations ciblées (robustesse, lisibilité, performance). C’est l’écart entre une réécriture aveugle du pipeline entier (coût LLM × N blocs, risque de régression) et une intervention chirurgicale sur le point de levier identifié par l’ablation. C’est le principe d’optimisation qui sous-tend le MLE-STAR : diagnostiquer avant d’agir.
class TargetedRefiner:
"""Raffine de maniere ciblee les composants critiques."""
def __init__(self, llm: LLMClient):
self.llm = llm
def refine_block(self, block: CodeBlock, task: str, current_score: float) -> str:
"""Genere une version amelioree d'un bloc."""
prompt = f"""Ameliore ce bloc de code ML pour la tache: {task}
BLOC: {block.name}
DESCRIPTION: {block.description}
CODE ACTUEL:
```python
{block.code[:1000]}
```
SCORE ACTUEL: {current_score:.3f}
Genere une version amelioree qui pourrait augmenter le score.
Focus sur les optimisations simples et efficaces.
```python
# Code ameliore
```"""
response = self.llm.generate(prompt, temperature=0.3)
match = re.search(r'```python\s*(.*?)\s*```', response, re.DOTALL)
return match.group(1).strip() if match else block.code
print("Classe TargetedRefiner definie : raffinement cible de blocs critiques via LLM")Classe TargetedRefiner definie : raffinement cible de blocs critiques via LLMclass MLEStarAblation:
"""Pipeline d'ablation complet de MLE-STAR."""
def __init__(self):
self.llm = LLMClient()
self.block_analyzer = CodeBlockAnalyzer(self.llm)
self.ablation_study = AblationStudy(self.llm)
self.refiner = TargetedRefiner(self.llm)
def analyze_and_refine(self, code: str, task: str, current_score: float = 0.5) -> Dict:
"""Analyse le code, identifie les blocs, et suggere des raffinements."""
print("[ANALYZER] Identification des blocs...")
blocks = self.block_analyzer.identify_blocks(code)
print(f" - {len(blocks)} blocs identifies")
if not blocks:
return {'success': False, 'error': 'Impossible d identifier les blocs'}
print("\n[ABLATION] Estimation de l'importance...")
blocks = self.ablation_study.estimate_importance(blocks, task)
print(" Resultats:")
for b in blocks[:3]:
print(f" - {b.name}: {b.importance:.2f}")
# Raffiner le bloc le plus important
top_block = blocks[0]
print(f"\n[REFINER] Raffinement de '{top_block.name}'...")
refined_code = self.refiner.refine_block(top_block, task, current_score)
return {
'success': True,
'blocks': [(b.name, b.importance) for b in blocks],
'top_block': top_block.name,
'refined_code': refined_code
}
print("Classe MLEStarAblation definie : pipeline complet Analyze -> Ablation -> Refine")Classe MLEStarAblation definie : pipeline complet Analyze -> Ablation -> RefinePourquoi tester le pipeline sur un exemple réaliste (RandomForest + feature engineering) : un exemple jouet (quelques lignes sans logique réelle) ne permettrait pas au CodeBlockAnalyzer d’identifier des blocs significatifs — l’ablation n’aurait rien à classer. L’exemple choisi (chargement CSV, feature engineering ratio/log, cross_val_score, RandomForest) est représentatif d’un pipeline ML standard avec ses quatre composants naturels (imports, data loading, training, évaluation). C’est ce réalisme qui donne à l’ablation une validité de face : les importances estimées seront cohérentes parce que l’exercice porte sur du code que l’on trouverait en production.
# Exemple de code ML a analyser
sample_code = '''
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score
# Feature Engineering
df = pd.read_csv('data.csv')
df['feature_ratio'] = df['feature_a'] / (df['feature_b'] + 1e-6)
df['feature_log'] = np.log1p(df['feature_c'])
X = df.drop('target', axis=1)
y = df['target']
# Model
model = RandomForestClassifier(n_estimators=100, random_state=42)
# Training & Evaluation
scores = cross_val_score(model, X, y, cv=5)
print(f"Accuracy: {scores.mean():.4f}")
'''
print("Code a analyser:")
print(sample_code[:300] + "...")Code a analyser:
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score
# Feature Engineering
df = pd.read_csv('data.csv')
df['feature_ratio'] = df['feature_a'] / (df['feature_b'] + 1e-6)
df['feature_log'] = np.log1p(df['feature_c'])
X = df.drop...Test du pipeline d’ablation sur un exemple de code ML.
# Test du pipeline d'ablation
pipeline = MLEStarAblation()
result = pipeline.analyze_and_refine(
code=sample_code,
task="binary classification tabular",
current_score=0.75
)
print("\n" + "="*50)
print("RESULTAT DE L'ABLATION:")
print("="*50)
print(f"\nBlocs identifies:")
for name, imp in result['blocks']:
print(f" - {name}: importance {imp:.2f}")
print(f"\nBloc prioritaire: {result['top_block']}")[ANALYZER] Identification des blocs...
- 4 blocs identifies
[ABLATION] Estimation de l'importance...
Resultats:
- Training & Evaluation: 0.90
- Feature Engineering: 0.80
- Model Definition: 0.70
[REFINER] Raffinement de 'Training & Evaluation'...
==================================================
RESULTAT DE L'ABLATION:
==================================================
Blocs identifies:
- Training & Evaluation: importance 0.90
- Feature Engineering: importance 0.80
- Model Definition: importance 0.70
- Import Libraries: importance 0.30
Bloc prioritaire: Training & EvaluationLecture des résultats d’ablation — le simulateur a identifié 4 blocs et estimé leur importance : « Training & Evaluation » = 0.90, « Feature Engineering » = 0.80, « Model Definition » = 0.70, « Import Libraries » = 0.30. Le bloc prioritaire retenu pour le raffinement est donc « Training & Evaluation ».
Ce que ce classement révèle sur la méthodologie : l’importance reflète le rôle sémantique de chaque bloc — l’entraînement et l’évaluation portent la performance mesurable du pipeline, tandis que les imports sont un prérequis mécanique sans valeur prédictive propre (0.30, dernier du classement). Ce verdict est une estimation experte du LLM, pas une ablation exécutée : il oriente l’effort de raffinement vers le bloc à plus fort levier sans payer le coût de N ré-entraînements. La rigueur consisterait à valider ce classement par une vraie ablation (ré-entraîner sans le bloc et mesurer la chute de performance) — mais pour la priorisation, l’estimation suffit à décider où agir.
Pourquoi ces libellés et ces scores ne sont pas des constantes. L’estimation est un appel LLM à température réduite (voir AblationStudy.estimate_importance) : relancer ce notebook change malgré tout les noms de blocs et leur ordre. Le même extrait de code a placé « Data Loading & Feature Engineering » en tête lors d’une exécution antérieure de ce lab, et « Training & Evaluation » dans l’exécution committée ici — les scores cités ci-dessus valent donc pour cette exécution, ils ne sont pas un résultat reproductible. C’est la raison de fond pour laquelle ce lab parle d’estimation et non d’ablation : un classement qui varie d’une exécution à l’autre ne se cite pas comme un fait, il se relit dans l’output. Les questions de réflexion en fin de lab (« Comment pourriez-vous valider l’importance expérimentalement ? ») trouvent ici leur réponse concrète — par la mesure, précisément parce que l’estimation est instable.
==================================================
CODE RAFFINE:
==================================================
from sklearn.model_selection import cross_val_score, StratifiedKFold, GridSearchCV
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import make_pipeline
from sklearn.ensemble import RandomForestClassifier
# Standardisation des données et utilisation d'un modèle de forêt aléatoire
pipeline = make_pipeline(StandardScaler(), RandomForestClassifier(random_state=42))
# Utilisation de StratifiedKFold pour une meilleure répartition des classes
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
# Hyperparameter tuning avec GridSearchCV
param_grid = {
'randomfo...Lecture du code raffiné — le TargetedRefiner a réécrit le bloc prioritaire (« Training & Evaluation », importance 0.90). Le bloc d’origine (validation croisée d’un RandomForestClassifier) revient sous la forme make_pipeline(StandardScaler(), RandomForestClassifier(...)) avec StratifiedKFold(shuffle=True) et un GridSearchCV : la standardisation et la stratification s’ajoutent à l’entraînement.
Ce que cet output démontre sur le raffinement ciblé : la sortie ne contient que le bloc prioritaire — refine_block est appelé sur un seul bloc, pas sur le pipeline entier. C’est la valeur ajoutée de la méthode MLE-STAR : au lieu d’une réécriture globale coûteuse et risquée (régressions possibles sur les blocs secondaires), l’intervention est chirurgicale, et l’effort est alloué là où l’ablation a montré qu’il aurait le plus d’impact. La contrepartie est visible dans le même output : le code rendu ne se suffit pas à lui-même — il ne porte plus le chargement des données ni le feature engineering de l’extrait d’origine, et le recoller au pipeline complet reste à la charge du praticien.
Appliquez l’étude d’ablation a un pipeline ML de votre choix pour identifier les composants a optimiser.
# TODO: Definissez votre propre pipeline ML
mon_pipeline = '''
# Exemple: classification, regression, ou clustering
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import GradientBoostingClassifier
# 1. Chargement des donnees
# 2. Preprocessing
# 3. Feature engineering
# 4. Entrainement modele
# 5. Evaluation
'''
# TODO: Analysez les blocs
pipeline = MLEStarAblation()
result = pipeline.analyze_and_refine(
code=mon_pipeline,
task="votre tache ici", # Ex: "multiclass classification"
current_score=0.70 # Score actuel hypothetique
)
# TODO: Affichez les blocs identifies
for name, importance in result['blocks']:
print(f"{name}: {importance:.2f}")
# TODO: Testez manuellement le code raffine
# Comparez avec une execution reelle si possible[ANALYZER] Identification des blocs...
- 5 blocs identifies
[ABLATION] Estimation de l'importance...
Resultats:
- Model Training: 0.90
- Data Loading: 0.80
- Model Evaluation: 0.80
[REFINER] Raffinement de 'Model Training'...
Model Training: 0.90
Data Loading: 0.80
Model Evaluation: 0.80
Data Preprocessing: 0.70
Feature Engineering: 0.60Créez un pipeline ML avec des blocs de feature engineering spécifiques et analysez comment l’ablation de chaque bloc de features affecte l’importance estimee par le LLM.
Indice : - Bloc 1: features originales (age, tenure, monthly_charges) - Bloc 2: features ratio (charges_per_month = total_charges / tenure) - Bloc 3: features polynomiales (age_squared, tenure_log) - Observez quel bloc le LLM estime le plus important et pourquoi
# Exercice : Ablation guidee sur le feature engineering
# Objectif : Comparer l'impact de differents blocs de features
# TODO: Definissez 3 pipelines avec des blocs de features differents
pipeline_original = """
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
# Features originales uniquement
df = pd.read_csv('churn.csv')
X = df[['age', 'tenure', 'monthly_charges', 'total_charges']]
y = df['churn']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)
score = model.score(X_test, y_test)
print(f"Accuracy: {score:.4f}")
"""
pipeline_avec_ratios = """
# TODO etudiant : ajoutez des features ratio
# Exemple: df['charges_per_month'] = df['total_charges'] / (df['tenure'] + 1)
# Exemple: df['charges_age_ratio'] = df['monthly_charges'] / df['age']
"""
pipeline_avec_polynomial = """
# TODO etudiant : ajoutez des features polynomiales
# Exemple: df['age_squared'] = df['age'] ** 2
# Exemple: df['tenure_log'] = np.log1p(df['tenure'])
"""
# TODO: Analysez chaque pipeline avec le pipeline d'ablation
# ablation = MLEStarAblation()
# for name, code in [("original", pipeline_original),
# ("ratios", pipeline_avec_ratios),
# ("polynomial", pipeline_avec_polynomial)]:
# result = ablation.analyze_and_refine(code, "customer churn prediction", current_score=0.75)
# print(f"\n--- {name} ---")
# for b_name, importance in result['blocks']:
# print(f" {b_name}: {importance:.2f}")
print("Exercice a completer : ablation guidee sur le feature engineering")Exercice a completer : ablation guidee sur le feature engineeringUtilisez l’ablation pour determiner la meilleure stratégie d’ensemble (voting, stacking, blending) pour un pipeline ML. L’objectif est de comparer les approches d’ensemble et de choisir la plus adaptee.
Indice : - Voting : VotingClassifier avec plusieurs modèles - Stacking : StackingClassifier avec meta-learner - Blending : predictions sur un validation set - L’ablation permet de determiner si l’ensemble apporte reellement un gain vs. un seul modèle
# Exercice : Comparaison de strategies d'ensemble par ablation
# Objectif : Determiner la meilleure strategie d'ensemble
# TODO: Definissez un pipeline avec strategie d'ensemble
ensemble_pipeline = """
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier, VotingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.preprocessing import StandardScaler
# Chargement et preparation
df = pd.read_csv('data.csv')
X = df.drop('target', axis=1)
y = df['target']
# Modeles individuels
rf = RandomForestClassifier(n_estimators=100, random_state=42)
gb = GradientBoostingClassifier(n_estimators=100, random_state=42)
lr = LogisticRegression(max_iter=1000)
# Strategie d'ensemble: Voting
ensemble = VotingClassifier(
estimators=[('rf', rf), ('gb', gb), ('lr', lr)],
voting='soft'
)
# Evaluation
scores = cross_val_score(ensemble, X, y, cv=5, scoring='roc_auc')
print(f"Ensemble AUC: {scores.mean():.4f} (+/- {scores.std():.4f})")
"""
# TODO: Analysez les blocs avec le pipeline d'ablation
# pipeline_ablation = MLEStarAblation()
# result = pipeline_ablation.analyze_and_refine(
# code=ensemble_pipeline,
# task="binary classification with ensemble",
# current_score=0.80
# )
#
# print("Blocs identifies:")
# for name, importance in result['blocks']:
# print(f" {name}: {importance:.2f}")
# print(f"\nBloc prioritaire: {result['top_block']}")
# TODO: Comparez avec un pipeline sans ensemble (modele unique)
# Quel est le gain estime de l'ensemble par rapport au modele seul ?
gain_estime = None # TODO etudiant : estimez le gain (ex: +0.02 AUC)
# TODO: Recommandez la meilleure strategie
# Justification : Voting est plus simple, Stacking plus performant mais plus couteux
recommandation = None # TODO etudiant : "voting" / "stacking" / "single_model"
print("Exercice a completer : comparaison de strategies d'ensemble par ablation")Exercice a completer : comparaison de strategies d'ensemble par ablation