Balanceo de carga entre modelos
Enrute los prompts sencillos a modelos pequeños y los complejos a modelos grandes.
Balanceo de carga entre modelos es una lección gratuita de AI Prompt Engineering en CoddyKit. Esta es la lección 3 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Prompt Engineering, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Prompt Engineering incluye 4 lecciones en total.
Por qué enrutar entre modelos
No todas las tareas necesitan el modelo más potente (y costoso). Una respuesta sencilla a un saludo no requiere GPT-4o. El enrutamiento de modelos dirige cada solicitud al modelo más barato capaz de gestionarla correctamente, lo que reduce los costes entre un 50 % y un 90 % y mantiene la calidad allí donde importa.
Enrutamiento basado en la complejidad
Clasifique la complejidad de la tarea antes de llamar a la API. Las tareas sencillas se envían a modelos económicos; las complejas, a modelos más capaces. Un clasificador ligero o unas heurísticas pueden tomar esta decisión rápidamente.
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)}')Clase de enrutador de modelos
Un enrutador de modelos asigna etiquetas de complejidad a modelos y dirige cada solicitud en consecuencia. La decisión de enrutamiento es determinista y se basa en umbrales 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}%)')Enrutamiento basado en los costes
Además de la complejidad, el enrutamiento basado en los costes tiene en cuenta el presupuesto de tokens, el nivel del usuario (gratuito o de pago) y los límites de gasto diarios para garantizar que los costes sean predecibles en todo el 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']Enrutamiento basado en la latencia
Los distintos modelos tienen perfiles de latencia diferentes. Cuando hay presión de tiempo (por ejemplo, en un chatbot con un SLA de 3 segundos), dirija las solicitudes a modelos más rápidos aunque sean menos capaces.
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'))Enrutamiento basado en las capacidades
Algunas tareas requieren capacidades específicas del modelo: visión, llamadas a funciones, contexto largo o un intérprete de código. El enrutamiento basado en capacidades garantiza que el modelo seleccionado pueda gestionar realmente la tarea.
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))Cadenas de fallback
Una cadena de fallback define el orden de los modelos que se deben probar cuando falla el modelo principal. Esto garantiza una alta disponibilidad incluso cuando proveedores concretos sufren interrupciones o problemas con los límites de tasa.
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')Registro de las decisiones de enrutamiento
Registre cada decisión de enrutamiento con suficiente contexto para auditarla, ajustar los umbrales y comprender la distribución de costes. Estos datos son esenciales para optimizar la lógica de enrutamiento con el tiempo.
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}')Pruebas A/B de modelos en producción
El enrutamiento de modelos también puede implementar pruebas A/B: dirigir un porcentaje del tráfico a un modelo nuevo para comparar la calidad antes de realizar el despliegue completo. Combine este proceso con la monitorización para tomar decisiones fundamentadas sobre la selección de modelos.
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}')Comprobaciones de estado de los modelos
Antes de dirigir tráfico a un modelo, verifique que responde correctamente. Una comprobación de estado envía un prompt conocido al modelo y valida la respuesta para confirmar que el proveedor 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')Análisis del impacto en los costes
Cuantifique el ahorro de costes obtenido mediante el enrutamiento de modelos. Con un tráfico compuesto por un 60 % de solicitudes SIMPLE, un 30 % de MODERATE y un 10 % de COMPLEX, un enrutamiento inteligente puede reducir los costes entre un 70 % y un 80 % frente a usar el mejor modelo para todo.
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()Comprobación rápida
Un usuario pregunta: 'Hi, how are you?' Su enrutador de modelos clasifica esto como SIMPLE. ¿Por qué es acertado dirigirlo a gpt-4o-mini en lugar de a gpt-4o?
Resumen del enrutamiento de modelos
El balanceo de carga entre modelos reduce los costes y garantiza que las capacidades se ajusten a las necesidades:
- Enrutamiento por complejidad: clasifica la tarea como SIMPLE/MODERATE/COMPLEX y la dirige al nivel de modelo correspondiente
- Enrutamiento basado en los costes: cambia a un modelo inferior cuando se agota el presupuesto o para usuarios del nivel gratuito
- Enrutamiento basado en la latencia: usa un modelo más rápido cuando el SLA es estricto
- Enrutamiento por capacidades: garantiza que el modelo seleccionado admita las funcionalidades necesarias (visión y llamadas a funciones)
- Cadenas de fallback: principal → fallback → emergencia para lograr una alta disponibilidad
- Pruebas A/B: prueba modelos nuevos con una parte del tráfico antes del despliegue completo
- Impacto en los costes: el enrutamiento puede reducir los costes entre un 70 % y un 80 % frente a usar siempre el mejor modelo
Preguntas frecuentes
¿La lección «Balanceo de carga entre modelos» es gratis?
Sí — el texto completo de «Balanceo de carga entre modelos» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Prompt Engineering, actualiza a CoddyKit PRO. El curso de AI Prompt Engineering incluye 4 lecciones en total.
¿Qué aprenderé en «Balanceo de carga entre modelos»?
Enrute los prompts sencillos a modelos pequeños y los complejos a modelos grandes. Practicas AI Prompt Engineering con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Prompt Engineering?
No se requiere experiencia previa. AI Prompt Engineering en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 3 de 4.
¿Cuánto tiempo toma la lección «Balanceo de carga entre modelos»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Prompt Engineering?
Sí. Cada lección de AI Prompt Engineering incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Estrategias de caché para prompts
- Procesamiento por lotes y ejecución asíncrona
- Balanceo de carga entre modelos
- Monitorización y alertas para pipelines de prompts