Balanceamento de carga entre modelos
Encaminhamento de prompts simples para modelos pequenos e de prompts complexos para modelos grandes.
Balanceamento de carga entre modelos é uma aula grátis de AI Prompt Engineering no CoddyKit. Esta é a aula 3 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Prompt Engineering, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Prompt Engineering inclui 4 aulas no total.
Por que fazer o roteamento entre modelos?
Nem toda tarefa precisa do modelo mais poderoso e caro. Uma resposta simples a uma saudação não exige o GPT-4o. O roteamento de modelos direciona cada solicitação ao modelo mais barato capaz de lidar bem com ela, reduzindo os custos em 50% a 90% e mantendo a qualidade onde ela é importante.
Roteamento baseado na complexidade
Classifique a complexidade da tarefa antes de chamar a API. Tarefas simples vão para modelos baratos; tarefas complexas vão para modelos capazes. Um classificador leve ou heurísticas podem tomar essa decisão rapidamente.
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 de roteamento de modelos
Um roteador de modelos associa rótulos de complexidade a modelos e direciona cada solicitação de acordo com essa associação. A decisão de roteamento é determinística e baseada em limites configuráveis.
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}%)')Roteamento consciente dos custos
Além da complexidade, o roteamento consciente dos custos considera o orçamento de tokens, o nível do usuário, gratuito ou pago, e os limites de gastos diários para garantir custos previsíveis em todo o 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']Roteamento consciente da latência
Modelos diferentes têm perfis de latência diferentes. Sob pressão de tempo, por exemplo, em um chatbot com um SLA de 3 segundos, direcione as solicitações para modelos mais rápidos, mesmo que sejam menos capazes.
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'))Roteamento consciente das capacidades
Algumas tarefas exigem capacidades específicas do modelo: visão, chamadas de funções, contexto longo ou interpretador de código. O roteamento consciente das capacidades garante que o modelo selecionado consiga realmente lidar com a tarefa.
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))Cadeias de alternativas
Uma cadeia de alternativas define a ordem dos modelos a serem tentados quando o modelo principal falha. Isso garante alta disponibilidade mesmo quando provedores individuais sofrem indisponibilidades ou problemas com limites de taxa.
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 das decisões de roteamento
Registre cada decisão de roteamento com contexto suficiente para auditá-la, ajustar os limites e entender a distribuição dos custos. Esses dados são essenciais para otimizar a lógica de roteamento ao longo do tempo.
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}')Testes A/B de modelos em produção
O roteamento de modelos também pode implementar testes A/B, direcionando uma porcentagem do tráfego para um modelo novo a fim de comparar a qualidade antes do lançamento geral. Combine isso com o monitoramento para tomar decisões baseadas em dados sobre a seleção 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}')Verificações de integridade dos modelos
Antes de direcionar tráfego para um modelo, verifique se ele está respondendo corretamente. Uma verificação de integridade envia uma instrução conhecida ao modelo e valida a resposta para confirmar que o provedor está disponível.
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álise do impacto nos custos
Quantifique a economia obtida com o roteamento de modelos. Com 60% de tráfego SIMPLE, 30% MODERATE e 10% COMPLEX, o roteamento inteligente pode reduzir os custos em 70% a 80% em comparação com o uso do melhor modelo para tudo.
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ção rápida
Um usuário pergunta: 'Oi, como você está?' Seu roteador de modelos classifica isso como SIMPLE. Por que direcionar essa solicitação para gpt-4o-mini em vez de gpt-4o é a decisão correta?
Resumo do roteamento de modelos
O balanceamento de carga entre modelos reduz os custos e garante compatibilidade com as capacidades necessárias:
- Roteamento por complexidade: classifique a tarefa como SIMPLE/MODERATE/COMPLEX e direcione-a para o nível de modelo correspondente
- Roteamento consciente dos custos: selecione um modelo mais barato quando o orçamento se esgotar ou para usuários do nível gratuito
- Roteamento consciente da latência: use um modelo mais rápido quando o SLA for rigoroso
- Roteamento por capacidades: garanta que o modelo selecionado ofereça suporte aos recursos necessários, como visão e chamadas de funções
- Cadeias de alternativas: principal → alternativa → emergência para alta disponibilidade
- Testes A/B: teste modelos novos em uma parcela do tráfego antes do lançamento geral
- Impacto nos custos: o roteamento pode reduzir os custos em 70% a 80% em comparação com usar sempre o melhor modelo
Perguntas Frequentes
A aula “Balanceamento de carga entre modelos” é grátis?
Sim — o texto completo de “Balanceamento de carga entre modelos” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Prompt Engineering, atualize para CoddyKit PRO. O curso de AI Prompt Engineering inclui 4 aulas no total.
O que vou aprender em “Balanceamento de carga entre modelos”?
Encaminhamento de prompts simples para modelos pequenos e de prompts complexos para modelos grandes. Você pratica AI Prompt Engineering com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.
Preciso ter experiência prévia para começar AI Prompt Engineering?
Nenhuma experiência prévia é necessária. AI Prompt Engineering no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 3 de 4.
Quanto tempo leva a aula “Balanceamento de carga entre modelos”?
A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.
Posso escrever e executar código nesta aula de AI Prompt Engineering?
Sim. Cada aula de AI Prompt Engineering inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.
Todas as aulas deste curso
- Estratégias de cache para prompts
- Processamento em lote e execução assíncrona
- Balanceamento de carga entre modelos
- Monitoramento e alertas para pipelines de prompts