Que fait Pipeline LLM Éco-Responsable ?
Modèles pour contrôler les coûts des API LLM tout en maintenant la qualité. Combine le routage des modèles, le suivi budgétaire, la logique de réessai et la mise en cache des prompts dans un pipeline modulaire.
Quand l'activer
- Construire des applications qui appellent des API LLM (Claude, GPT, etc.)
- Traiter des lots d'éléments de complexité variable
- Devoir respecter un budget pour les dépenses API
- Optimiser les coûts sans sacrifier la qualité des tâches complexes
Concepts Fondamentaux
1. Routage des Modèles par Complexité de la Tâche
Sélectionner automatiquement des modèles moins chers pour les tâches simples, en réservant les modèles coûteux pour les tâches complexes.
MODEL_SONNET = "claude-sonnet-5"
MODEL_HAIKU = "claude-haiku-4-5-20251001"
_SEUIL_TEXTE_SONNET = 10_000 # caractères
_SEUIL_ELEMENTS_SONNET = 30 # éléments
def selectionner_modele(
longueur_texte: int,
nombre_elements: int,
modele_force: str | None = None,
) -> str:
"""Sélectionner le modèle en fonction de la complexité de la tâche."""
if modele_force is not None:
return modele_force
if longueur_texte >= _SEUIL_TEXTE_SONNET or nombre_elements >= _SEUIL_ELEMENTS_SONNET:
return MODEL_SONNET # Tâche complexe
return MODEL_HAIKU # Tâche simple (3-4x moins cher)
2. Suivi Immutable des Coûts
Suivre les dépenses cumulées avec des dataclasses figées. Chaque appel API renvoie un nouveau tracker — il ne mute jamais l'état.
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class EnregistrementCout:
modele: str
jetons_entree: int
jetons_sortie: int
cout_usd: float
@dataclass(frozen=True, slots=True)
class TrackerCout:
limite_budget: float = 1.00
enregistrements: tuple[EnregistrementCout, ...] = ()
def ajouter(self, enregistrement: EnregistrementCout) -> "TrackerCout":
"""Renvoie un nouveau tracker avec l'enregistrement ajouté (ne mute jamais self)."""
return TrackerCout(
limite_budget=self.limite_budget,
enregistrements=(*self.enregistrements, enregistrement),
)
@property
def cout_total(self) -> float:
return sum(r.cout_usd for r in self.enregistrements)
@property
def hors_budget(self) -> bool:
return self.cout_total > self.limite_budget
3. Logique de Réessai Restrictive
Ne réessayer que sur les erreurs transitoires. Échouer rapidement sur les erreurs d'authentification ou de requête incorrecte.
from anthropic import (
APIConnectionError,
InternalServerError,
RateLimitError,
)
_ERREURS_REESSAYABLES = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_REESSAIS = 3
def appeler_avec_reessai(func, *, max_reessais: int = _MAX_REESSAIS):
"""Ne réessayer que sur les erreurs transitoires, échouer rapidement sur les autres."""
for tentative in range(max_reessais):
try:
return func()
except _ERREURS_REESSAYABLES:
if tentative == max_reessais - 1:
raise
time.sleep(2 ** tentative) # Backoff exponentiel
# AuthenticationError, BadRequestError etc. → lever immédiatement
4. Mise en Cache des Prompts
Mettre en cache les prompts système longs pour éviter de les renvoyer à chaque requête.
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt_systeme,
"cache_control": {"type": "ephemeral"}, # Mettre en cache ceci
},
{
"type": "text",
"text": entree_utilisateur, # Partie variable
},
],
}
]
Composition
Combiner les quatre techniques dans une seule fonction pipeline :
def traiter(texte: str, config: Config, tracker: TrackerCout) -> tuple[Resultat, TrackerCout]:
# 1. Router le modèle
modele = selectionner_modele(len(texte), elements_estimes, config.modele_force)
# 2. Vérifier le budget
if tracker.hors_budget:
raise BudgetDepasseErreur(tracker.cout_total, tracker.limite_budget)
# 3. Appeler avec réessai + mise en cache
reponse = appeler_avec_reessai(lambda: client.messages.create(
modele=modele,
messages=construire_messages_mis_en_cache(prompt_systeme, texte),
))
# 4. Suivre le coût (immuable)
enregistrement = EnregistrementCout(modele=modele, jetons_entree=..., jetons_sortie=..., cout_usd=...)
tracker = tracker.ajouter(enregistrement)
return analyser_resultat(reponse), tracker
Référence de Prix (2026)
| Modèle | Entrée ($/1M jetons) | Sortie ($/1M jetons) | Coût Relatif |
|---|---|---|---|
| Haiku 3.5 (legacy) | 0,80 $ | 4,00 $ | 0,8x |
| Haiku 4.5 | 1,00 $ | 5,00 $ | 1x |
| Sonnet 5 | 2,00 $ | 10,00 $ | 2x |
| Sonnet 4.6 | 3,00 $ | 15,00 $ | 3x |
| Opus 4.8 | 5,00 $ | 25,00 $ | 5x |
| Fable 5 / Mythos 5 | 10,00 $ | 50,00 $ | 10x |
| Opus 4.0 / 4.1 (legacy) | 15,00 $ | 75,00 $ | 15x |
Bonnes Pratiques
- Commencez avec le modèle le moins cher et ne routez vers des modèles coûteux que lorsque les seuils de complexité sont atteints
- Définissez des limites budgétaires explicites avant de traiter des lots — échouez tôt plutôt que de dépasser le budget
- Journalisez les décisions de sélection du modèle afin de pouvoir ajuster les seuils en fonction des données réelles
- Utilisez la mise en cache des prompts pour les prompts système de plus de 1024 jetons — économise coût et latence
- Ne réessayez jamais sur les erreurs d'authentification ou de validation — uniquement sur les échecs transitoires (réseau, limite de débit, erreur serveur)
Anti-Modèles à Éviter
- Utiliser le modèle le plus cher pour toutes les requêtes, quelle que soit la complexité
- Réessayer sur toutes les erreurs (gaspille le budget sur des échecs permanents)
- Muter l'état de suivi des coûts (rend le débogage et l'audit difficiles)
- Coder en dur les noms de modèles dans la base de code (utilisez des constantes ou une configuration)
- Ignorer la mise en cache des prompts pour les prompts système répétitifs
Quand l'Utiliser
- Toute application appelant les API LLM de Claude, OpenAI ou similaires
- Pipelines de traitement par lots où les coûts s'accumulent rapidement
- Architectures multi-modèles nécessitant un routage intelligent
- Systèmes de production nécessitant des garde-fous budgétaires