Bilanciamento del carico tra modelli
Instradamento dei prompt semplici verso modelli piccoli e di quelli complessi verso modelli grandi
Bilanciamento del carico tra modelli è una lezione AI Prompt Engineering gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Prompt Engineering, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Prompt Engineering include 4 lezioni in totale.
Perché distribuire le richieste tra i modelli
Non tutte le attività richiedono il modello più potente (e costoso). Una semplice risposta di saluto non richiede GPT-4o. Il routing dei modelli indirizza ogni richiesta verso il modello meno costoso in grado di gestirla adeguatamente, riducendo i costi del 50-90% e mantenendo la qualità dove conta.
Routing basato sulla complessità
Classifichi la complessità dell'attività prima di chiamare l'API. Le attività semplici vengono indirizzate verso modelli economici, quelle complesse verso modelli più capaci. Un classificatore leggero o alcune euristiche possono prendere rapidamente questa decisione.
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 Model Router
Un model router associa le etichette di complessità ai modelli e indirizza ogni richiesta di conseguenza. La decisione di routing è deterministica e si basa su soglie configurabili.
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}%)')Routing consapevole dei costi
Oltre alla complessità, il routing consapevole dei costi considera il budget di token, il livello dell'utente (gratuito o a pagamento) e i limiti di spesa giornalieri, per garantire costi prevedibili in tutto il sistema.
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']Routing consapevole della latenza
I diversi modelli presentano profili di latenza differenti. In caso di pressione sui tempi (ad esempio, per un chatbot con uno SLA di 3 secondi), indirizzi le richieste verso modelli più veloci anche se meno capaci.
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'))Routing consapevole delle capacità
Alcune attività richiedono capacità specifiche del modello: visione, function calling, contesto esteso o interprete di codice. Il routing consapevole delle capacità garantisce che il modello selezionato sia effettivamente in grado di gestire l'attività.
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))Catene di fallback
Una catena di fallback definisce l'ordine dei modelli da provare quando il modello principale non funziona. In questo modo si garantisce un'elevata disponibilità anche quando singoli provider subiscono interruzioni o problemi legati ai limiti di frequenza.
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')Registrazione delle decisioni di routing
Registri ogni decisione di routing con un contesto sufficiente per eseguire audit, ottimizzare le soglie e comprendere la distribuzione dei costi. Questi dati sono essenziali per ottimizzare nel tempo la logica di routing.
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}')Test A/B dei modelli in produzione
Il routing dei modelli può anche implementare test A/B, indirizzando una percentuale del traffico verso un nuovo modello per confrontarne la qualità prima del rilascio completo. Combini questa strategia con il monitoraggio per prendere decisioni basate sui dati nella selezione dei modelli.
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}')Controlli dello stato dei modelli
Prima di indirizzare il traffico verso un modello, verifichi che risponda correttamente. Un controllo dello stato interroga il modello con un prompt noto e convalida la risposta per confermare che il provider sia disponibile.
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')Analisi dell'impatto sui costi
Quantifichi il risparmio ottenuto grazie al routing dei modelli. Con un traffico composto per il 60% da richieste SIMPLE, per il 30% da MODERATE e per il 10% da COMPLEX, un routing intelligente può ridurre i costi del 70-80% rispetto all'utilizzo del modello migliore per ogni richiesta.
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()Verifica rapida
Un utente chiede: 'Salve, come sta?'. Il model router classifica la richiesta come SIMPLE. Perché indirizzarla a gpt-4o-mini invece che a gpt-4o è la decisione corretta?
Riepilogo del routing dei modelli
Il bilanciamento del carico tra i modelli riduce i costi e garantisce la corrispondenza tra attività e capacità:
- Routing basato sulla complessità: classifichi l'attività come SIMPLE/MODERATE/COMPLEX e la indirizzi al livello di modello corrispondente
- Routing consapevole dei costi: riduca il livello del modello quando si esaurisce il budget o per gli utenti del livello gratuito
- Routing consapevole della latenza: usi un modello più veloce quando lo SLA è stringente
- Routing basato sulle capacità: garantisca che il modello selezionato supporti le funzionalità richieste (visione, function calling)
- Catene di fallback: principale → fallback → emergenza per un'elevata disponibilità
- Test A/B: testi i nuovi modelli su una porzione del traffico prima del rilascio completo
- Impatto sui costi: il routing può ridurre i costi del 70-80% rispetto all'utilizzo costante del modello migliore
Domande Frequenti
La lezione «Bilanciamento del carico tra modelli» è gratuita?
Sì — il testo completo di «Bilanciamento del carico tra modelli» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Prompt Engineering, passa a CoddyKit PRO. Il corso AI Prompt Engineering include 4 lezioni in totale.
Cosa imparerò in «Bilanciamento del carico tra modelli»?
Instradamento dei prompt semplici verso modelli piccoli e di quelli complessi verso modelli grandi Eserciti AI Prompt Engineering con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Prompt Engineering?
Non è richiesta alcuna esperienza precedente. AI Prompt Engineering su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Bilanciamento del carico tra modelli»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Prompt Engineering?
Sì. Ogni lezione AI Prompt Engineering include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Strategie di caching per i prompt
- Elaborazione batch ed esecuzione asincrona
- Bilanciamento del carico tra modelli
- Monitoraggio e alerting per le pipeline di prompt