Équilibrage de charge entre les modèles
Acheminez les prompts simples vers de petits modèles et les prompts complexes vers de grands modèles.
Équilibrage de charge entre les modèles est une leçon AI Prompt Engineering gratuite sur CoddyKit. Ceci est la leçon 3 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Prompt Engineering, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Prompt Engineering comprend 4 leçons au total.
Pourquoi répartir les requêtes entre plusieurs modèles ?
Toutes les tâches n'ont pas besoin du modèle le plus puissant, et donc le plus coûteux. Une simple réponse de salutation ne nécessite pas GPT-4o. Le routage des modèles dirige chaque requête vers le modèle le moins coûteux capable de la traiter correctement, ce qui réduit les coûts de 50 à 90 % tout en maintenant la qualité là où elle compte.
Routage fondé sur la complexité
Classez la complexité de la tâche avant d'appeler l'API. Les tâches simples sont dirigées vers des modèles peu coûteux ; les tâches complexes vers des modèles plus performants. Un classificateur léger ou des heuristiques peuvent prendre rapidement cette décision.
COMPLEXITY_CLASSIFIER_PROMPT = '''Classify the complexity of this user request.
Return ONLY one word: SIMPLE, MODERATE, or COMPLEX.
SIMPLE: greeting, factual lookup, single-step question, direct answer needed
MODERATE: multi-step explanation, comparison, short analysis, code snippet
COMPLEX: deep analysis, long code generation, reasoning chain, specialized domain
Request: {request}'''
import openai
client_mini = openai.OpenAI(api_key='YOUR_API_KEY')
def classify_complexity(request):
response = client_mini.chat.completions.create(
model='gpt-4o-mini', # always use cheap model for classifier
messages=[{'role': 'user', 'content':
COMPLEXITY_CLASSIFIER_PROMPT.format(request=request)}],
max_tokens=5,
temperature=0
)
label = response.choices[0].message.content.strip().upper()
if label not in ('SIMPLE', 'MODERATE', 'COMPLEX'):
label = 'MODERATE' # safe default
return label
for req in ['Hi', 'Explain quicksort', 'Design a distributed systems architecture']:
print(f'{req[:40]}: {classify_complexity(req)}')Classe ModelRouter
Un routeur de modèles associe des étiquettes de complexité à des modèles et dirige chaque requête en conséquence. La décision de routage est déterministe et repose sur des seuils configurables.
MODEL_ROUTES = {
'SIMPLE': {
'model': 'gpt-4o-mini',
'max_tokens': 300,
'cost_per_1k_input': 0.00015,
'use_case': 'greetings, FAQ, simple factual questions'
},
'MODERATE': {
'model': 'gpt-4o',
'max_tokens': 1500,
'cost_per_1k_input': 0.0025,
'use_case': 'explanations, analysis, code snippets'
},
'COMPLEX': {
'model': 'claude-opus-4-5',
'max_tokens': 4096,
'cost_per_1k_input': 0.015,
'use_case': 'deep reasoning, long code, specialized domains'
}
}
class ModelRouter:
def __init__(self):
self.routes = MODEL_ROUTES
self.call_counts = {k: 0 for k in MODEL_ROUTES}
def route(self, request, messages):
complexity = classify_complexity(request)
route = self.routes[complexity]
self.call_counts[complexity] += 1
print(f'Routing "{request[:40]}" -> {route["model"]} ({complexity})')
return route['model'], route['max_tokens']
def cost_report(self):
total = sum(self.call_counts.values())
for complexity, count in self.call_counts.items():
pct = count / total * 100 if total else 0
print(f'{complexity}: {count} calls ({pct:.0f}%)')Routage tenant compte des coûts
Au-delà de la complexité, le routage tenant compte des coûts prend en considération le budget de jetons, le niveau de l'utilisateur, gratuit ou payant, et les plafonds de dépenses quotidiennes afin de garantir des coûts prévisibles pour l'ensemble du système.
class CostAwareRouter(ModelRouter):
def __init__(self, daily_budget_usd=100.0):
super().__init__()
self.daily_budget = daily_budget_usd
self.daily_spent = 0.0
def estimate_cost(self, model, input_tokens, max_output_tokens):
route = next((r for r in self.routes.values() if r['model'] == model), None)
if not route:
return 0.0
return (
(input_tokens / 1000) * route['cost_per_1k_input'] +
(max_output_tokens / 1000) * route['cost_per_1k_input'] * 3
)
def route_with_budget(self, request, messages, user_tier='free'):
complexity = classify_complexity(request)
# Downgrade if budget is exhausted or user is on free tier
budget_remaining = self.daily_budget - self.daily_spent
if budget_remaining < 0.01 or user_tier == 'free':
complexity = 'SIMPLE' # downgrade to cheapest model
print('Budget constraint: routing to SIMPLE model')
route = self.routes[complexity]
input_tokens = sum(len(m['content'].split()) for m in messages) * 1.3
cost = self.estimate_cost(route['model'], input_tokens, route['max_tokens'])
self.daily_spent += cost
return route['model'], route['max_tokens']Routage tenant compte de la latence
Les différents modèles ont des profils de latence différents. En cas de contrainte de temps, par exemple pour un agent conversationnel avec un SLA de 3 secondes, dirigez les requêtes vers des modèles plus rapides, même s'ils sont moins performants.
import time
# Model latency profiles (approximate P95 values)
MODEL_LATENCY_P95 = {
'gpt-4o-mini': 1.5, # seconds
'gpt-4o': 4.0,
'claude-haiku-4-5': 1.2,
'claude-sonnet-4-5': 3.0,
'claude-opus-4-5': 6.0
}
SLA_LATENCY_BUDGET = 3.0 # seconds
def route_with_latency_constraint(complexity, sla_seconds=SLA_LATENCY_BUDGET):
route = MODEL_ROUTES[complexity]
p95_latency = MODEL_LATENCY_P95.get(route['model'], 5.0)
if p95_latency > sla_seconds:
# Find fastest model under SLA
affordable_models = [
(lat, m) for m, lat in MODEL_LATENCY_P95.items()
if lat <= sla_seconds
]
if affordable_models:
fastest = min(affordable_models)[1]
print(f'Latency constraint: downgrading from {route["model"]} to {fastest}')
return fastest
return route['model']
print('COMPLEX request under 3s SLA:', route_with_latency_constraint('COMPLEX'))Routage tenant compte des capacités
Certaines tâches nécessitent des capacités précises du modèle : vision, appel de fonctions, contexte étendu ou interpréteur de code. Le routage tenant compte des capacités garantit que le modèle sélectionné peut réellement traiter la tâche.
MODEL_CAPABILITIES = {
'gpt-4o-mini': {
'vision': True,
'function_calling': True,
'context_window': 128000,
'code_interpreter': False
},
'gpt-4o': {
'vision': True,
'function_calling': True,
'context_window': 128000,
'code_interpreter': True
},
'claude-opus-4-5': {
'vision': True,
'function_calling': True,
'context_window': 200000,
'code_interpreter': False
}
}
def capability_aware_route(required_capabilities, context_length=0):
candidates = []
for model, caps in MODEL_CAPABILITIES.items():
if context_length > caps['context_window']:
continue
if all(caps.get(cap, False) for cap in required_capabilities):
candidates.append(model)
if not candidates:
raise ValueError(f'No model supports: {required_capabilities}')
# Among capable models, pick cheapest
cost_rank = ['gpt-4o-mini', 'claude-opus-4-5', 'gpt-4o']
for model in cost_rank:
if model in candidates:
return model
return candidates[0]
print(capability_aware_route(['vision', 'function_calling'], context_length=5000))Chaînes de secours
Une chaîne de secours définit l'ordre des modèles à essayer lorsque le modèle principal échoue. Elle garantit une haute disponibilité, même lorsque certains fournisseurs rencontrent des pannes ou des problèmes de limitation de débit.
FALLBACK_CHAINS = {
'primary': 'claude-opus-4-5',
'fallback': 'gpt-4o',
'emergency': 'gpt-4o-mini'
}
def call_with_fallback(messages, chain=FALLBACK_CHAINS):
providers = [
('anthropic', chain['primary']),
('openai', chain['fallback']),
('openai', chain['emergency'])
]
for provider, model in providers:
try:
print(f'Trying {model}...')
if provider == 'anthropic':
import anthropic
ac = anthropic.Anthropic(api_key='YOUR_KEY')
resp = ac.messages.create(
model=model, max_tokens=500, messages=messages
)
return resp.content[0].text
else:
import openai
oc = openai.OpenAI(api_key='YOUR_KEY')
resp = oc.chat.completions.create(
model=model, messages=messages, max_tokens=500
)
return resp.choices[0].message.content
except Exception as e:
print(f'{model} failed: {e}. Trying next...')
raise RuntimeError('All models in fallback chain failed')Journalisation des décisions de routage
Journalisez chaque décision de routage avec suffisamment de contexte pour pouvoir l'auditer, ajuster les seuils et comprendre la répartition des coûts. Ces données sont essentielles pour optimiser progressivement la logique de routage.
import json
from datetime import datetime
ROUTING_LOG_FILE = 'routing_decisions.jsonl'
def log_routing_decision(request_id, request_text, complexity,
model_selected, cost_estimate, latency_ms,
user_tier='free'):
entry = {
'timestamp': datetime.utcnow().isoformat(),
'request_id': request_id,
'request_preview': request_text[:50],
'complexity': complexity,
'model': model_selected,
'cost_estimate_usd': round(cost_estimate, 6),
'latency_ms': round(latency_ms),
'user_tier': user_tier
}
with open(ROUTING_LOG_FILE, 'a') as f:
f.write(json.dumps(entry) + '\n')
# Analyze routing log to tune thresholds
def analyze_routing_log():
from collections import Counter
model_counts = Counter()
total_cost = 0.0
with open(ROUTING_LOG_FILE) as f:
for line in f:
e = json.loads(line)
model_counts[e['model']] += 1
total_cost += e['cost_estimate_usd']
print('Model distribution:', dict(model_counts))
print(f'Total estimated cost: ${total_cost:.4f}')Tests A/B de modèles en production
Le routage des modèles peut également mettre en œuvre des tests A/B : dirigez un pourcentage du trafic vers un nouveau modèle afin de comparer sa qualité avant un déploiement complet. Associez cette pratique à une surveillance pour prendre des décisions de sélection de modèles fondées sur les données.
import random
class ABModelRouter:
def __init__(self, control_model, treatment_model, treatment_pct=10):
self.control = control_model
self.treatment = treatment_model
self.treatment_pct = treatment_pct
self.assignment_log = {} # request_id: 'control' | 'treatment'
def route(self, request_id):
if request_id in self.assignment_log:
# Sticky assignment: same user always gets same model
return self.assignment_log[request_id]
if random.random() * 100 < self.treatment_pct:
assignment = 'treatment'
model = self.treatment
else:
assignment = 'control'
model = self.control
self.assignment_log[request_id] = assignment
return model, assignment
# Usage
ab_router = ABModelRouter(
control_model='gpt-4o',
treatment_model='claude-opus-4-5',
treatment_pct=10 # 10% get new model
)
for user_id in range(5):
result = ab_router.route(f'user_{user_id}')
print(f'user_{user_id}: {result}')Vérifications de l'état des modèles
Avant de diriger du trafic vers un modèle, vérifiez qu'il répond correctement. Une vérification de l'état envoie au modèle une instruction connue et valide la réponse pour confirmer que le fournisseur est disponible.
import time
def health_check(model, provider='openai', timeout=5):
'''
Returns True if model is healthy, False if timed out or errored.
'''
test_prompt = 'Reply with exactly: OK'
try:
start = time.time()
if provider == 'openai':
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
resp = client.chat.completions.create(
model=model,
messages=[{'role': 'user', 'content': test_prompt}],
max_tokens=5,
timeout=timeout
)
text = resp.choices[0].message.content.strip()
elif provider == 'anthropic':
import anthropic
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
resp = client.messages.create(
model=model, max_tokens=5,
messages=[{'role': 'user', 'content': test_prompt}],
)
text = resp.content[0].text.strip()
latency = (time.time() - start) * 1000
healthy = 'ok' in text.lower()
print(f'{model}: {"HEALTHY" if healthy else "DEGRADED"} ({latency:.0f}ms)')
return healthy
except Exception as e:
print(f'{model}: UNHEALTHY ({e})')
return False
# Run health checks before routing critical traffic
# health_check('gpt-4o-mini', provider='openai')
# health_check('claude-haiku-4-5', provider='anthropic')Analyse de l'impact sur les coûts
Quantifiez les économies réalisées grâce au routage des modèles. Avec un trafic composé de 60 % de tâches SIMPLE, 30 % de tâches MODERATE et 10 % de tâches COMPLEX, un routage intelligent peut réduire les coûts de 70 à 80 % par rapport à l'utilisation du meilleur modèle pour toutes les tâches.
def cost_impact_analysis(daily_requests=10000):
# Traffic distribution
traffic = {'SIMPLE': 0.60, 'MODERATE': 0.30, 'COMPLEX': 0.10}
# Avg tokens per request (input + output)
avg_tokens = {'SIMPLE': 500, 'MODERATE': 2000, 'COMPLEX': 5000}
# Pricing per 1K tokens (blended input+output)
pricing = {'SIMPLE': 0.00030, 'MODERATE': 0.01000, 'COMPLEX': 0.04500}
premium_price = 0.04500 # if we used COMPLEX model for everything
routed_cost = 0.0
premium_cost = 0.0
for complexity, pct in traffic.items():
requests = daily_requests * pct
tokens = avg_tokens[complexity]
routed_cost += requests * (tokens / 1000) * pricing[complexity]
premium_cost += requests * (tokens / 1000) * premium_price
savings_pct = (1 - routed_cost / premium_cost) * 100
print(f'Daily requests: {daily_requests:,}')
print(f'With routing: ${routed_cost:,.2f}/day')
print(f'Without routing: ${premium_cost:,.2f}/day')
print(f'Savings: {savings_pct:.0f}% (${premium_cost - routed_cost:,.2f}/day)')
cost_impact_analysis()Vérification rapide
Un utilisateur demande : « Bonjour, comment allez-vous ? » Votre routeur de modèles classe cette demande comme SIMPLE. Pourquoi est-il préférable de la diriger vers gpt-4o-mini plutôt que vers gpt-4o ?
Résumé du routage des modèles
La répartition de charge entre les modèles réduit les coûts et garantit l'adéquation des capacités :
- Routage par complexité : classer la tâche comme SIMPLE/MODERATE/COMPLEX et la diriger vers le niveau de modèle correspondant
- Routage tenant compte des coûts : choisir un modèle moins coûteux lorsque le budget est épuisé ou pour un utilisateur gratuit
- Routage tenant compte de la latence : utiliser un modèle plus rapide lorsque le SLA est strict
- Routage par capacités : vérifier que le modèle sélectionné prend en charge les fonctionnalités requises, comme la vision et les appels de fonctions
- Chaînes de secours : modèle principal → modèle de secours → modèle d'urgence pour garantir une haute disponibilité
- Tests A/B : tester de nouveaux modèles sur une partie du trafic avant le déploiement complet
- Impact sur les coûts : le routage peut réduire les coûts de 70 à 80 % par rapport à l'utilisation systématique du meilleur modèle
Questions Fréquemment Posées
La leçon « Équilibrage de charge entre les modèles » est-elle gratuite ?
Oui — le texte complet de « Équilibrage de charge entre les modèles » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Prompt Engineering, passe à CoddyKit PRO. Le cours AI Prompt Engineering comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Équilibrage de charge entre les modèles » ?
Acheminez les prompts simples vers de petits modèles et les prompts complexes vers de grands modèles. Tu pratiques AI Prompt Engineering avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer AI Prompt Engineering ?
Aucune expérience préalable n'est requise. AI Prompt Engineering sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 3 sur 4.
Combien de temps prend la leçon « Équilibrage de charge entre les modèles » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon AI Prompt Engineering ?
Oui. Chaque leçon AI Prompt Engineering inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Stratégies de mise en cache des prompts
- Traitement par lots et exécution asynchrone
- Équilibrage de charge entre les modèles
- Surveillance et alertes pour les pipelines de prompts