Que fait Pipeline LLM à Coût Maîtrisé ?
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 nouvelle tentative et la mise en cache des prompts en un pipeline composable.
Quand 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 le coût sans sacrifier la qualité sur les tâches complexes
Concepts Fondamentaux
1. Routage des Modèles par Complexité de la Tâche
Sélectionner automatiquement les modèles 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"
_SONNET_TEXT_THRESHOLD = 10_000 # caractères
_SONNET_ITEM_THRESHOLD = 30 # éléments
def select_model(
text_length: int,
item_count: int,
force_model: str | None = None,
) -> str:
"""Sélectionne le modèle en fonction de la complexité de la tâche."""
if force_model is not None:
return force_model
if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:
return MODEL_SONNET # Tâche complexe
return MODEL_HAIKU # Tâche simple (3-4x moins cher)
2. Suivi des Coûts Immuable
Suivre les dépenses cumulées avec des dataclasses figées. Chaque appel API retourne un nouveau tracker — ne mute jamais l'état.
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class CostRecord:
model: str
input_tokens: int
output_tokens: int
cost_usd: float
@dataclass(frozen=True, slots=True)
class CostTracker:
budget_limit: float = 1.00
records: tuple[CostRecord, ...] = ()
def add(self, record: CostRecord) -> "CostTracker":
"""Retourne un nouveau tracker avec l'enregistrement ajouté (ne mute jamais self)."""
return CostTracker(
budget_limit=self.budget_limit,
records=(*self.records, record),
)
@property
def total_cost(self) -> float:
return sum(r.cost_usd for r in self.records)
@property
def over_budget(self) -> bool:
return self.total_cost > self.budget_limit
3. Logique de Nouvelle Tentative Restreinte
Ne réessayer que sur les erreurs transitoires. Échouer rapidement sur les erreurs d'authentification ou de requête invalide.
from anthropic import (
APIConnectionError,
InternalServerError,
RateLimitError,
)
_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_RETRIES = 3
def call_with_retry(func, *, max_retries: int = _MAX_RETRIES):
"""Ne réessaie que sur les erreurs transitoires, échoue rapidement sur les autres."""
for attempt in range(max_retries):
try:
return func()
except _RETRYABLE_ERRORS:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 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": system_prompt,
"cache_control": {"type": "ephemeral"}, # Mettre en cache ceci
},
{
"type": "text",
"text": user_input, # Partie variable
},
],
}
]
Composition
Combiner les quatre techniques dans une seule fonction pipeline :
def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:
# 1. Router le modèle
model = select_model(len(text), estimated_items, config.force_model)
# 2. Vérifier le budget
if tracker.over_budget:
raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)
# 3. Appeler avec nouvelle tentative + mise en cache
response = call_with_retry(lambda: client.messages.create(
model=model,
messages=build_cached_messages(system_prompt, text),
))
# 4. Suivre le coût (immuable)
record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)
tracker = tracker.add(record)
return parse_result(response), tracker
Référence des Prix (2026)
| Modèle | Entrée ($/M tokens) | Sortie ($/M tokens) | 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
- Commencer par le modèle le moins cher et ne router vers les modèles coûteux que lorsque les seuils de complexité sont atteints
- Définir des limites budgétaires explicites avant de traiter les lots — échouer tôt plutôt que de dépasser le budget
- Journaliser les décisions de sélection du modèle afin de pouvoir ajuster les seuils sur la base de données réelles
- Utiliser la mise en cache des prompts pour les prompts système de plus de 1024 tokens — économise à la fois le coût et la latence
- Ne jamais réessayer sur les erreurs d'authentification ou de validation — uniquement les défaillances transitoires (réseau, limite de débit, erreur serveur)
Anti-Patrons à Éviter
- Utiliser le modèle le plus cher pour toutes les requêtes, indépendamment de la complexité
- Réessayer sur toutes les erreurs (gaspille le budget sur les défaillances permanentes)
- 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 (utiliser des constantes ou une configuration)
- Ignorer la mise en cache des prompts pour les prompts système répétitifs
Quand Utiliser
- Toute application appelant Claude, OpenAI ou des API LLM 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