Módulo 4: Middleware y Customización
Dynamic Models: Selección Inteligente
Descripción de la cápsula
En las cápsulas anteriores aprendiste a interceptar llamadas al modelo con @wrap_model_call y a personalizar tools con @wrap_tool_call. Pero hasta ahora, siempre usaste un solo modelo para todo. Eso es como usar un camión de carga para ir a comprar pan: funciona, pero es ineficiente.
Uno de los patrones más poderosos del middleware system es model routing: el agente selecciona automáticamente qué modelo usar basándose en la complejidad de cada request. Preguntas simples como "¿Qué hora es?" van a un modelo rápido y económico (gpt-4.1-mini). Preguntas complejas como "Analiza las diferencias entre 5 frameworks de IA" van a un modelo potente (gpt-4.1 o claude-sonnet). Esto te permite reducir costos hasta un 80% sin sacrificar calidad en las respuestas que realmente lo necesitan.
La clave está en @wrap_model_call: ya sabes que intercepta la llamada al modelo — ahora vas a usarlo para cambiar el modelo mismo según la clasificación de la pregunta. Combinado con init_chat_model, puedes pre-inicializar múltiples modelos y switchear entre ellos de forma transparente.
El concepto: routing de modelos por complejidad
En un sistema de producción, no todas las preguntas son iguales:
| Tipo de pregunta | Ejemplo | Modelo ideal |
|---|---|---|
| Saludo / trivial | "Hola, ¿cómo estás?" | gpt-4.1-mini (~$0.0004/call) |
| Factual simple | "¿Cuál es la capital de Francia?" | gpt-4.1-mini |
| Multi-step | "Compara React, Vue y Angular" | gpt-4.1 (~$0.005/call) |
| Análisis profundo | "Investiga y analiza tendencias de IA en 2025" | gpt-4.1 o claude-sonnet |
Usar gpt-4.1 para todo funciona, pero si el 70% de tus requests son simples, estás pagando 10x más de lo necesario en esas consultas. Model routing automatiza esta decisión.
Arquitectura del routing
Usuario envía pregunta
│
▼
┌─────────────────────┐
│ classify_complexity │ ← Heurísticas: longitud, keywords, tools
│ "simple" / "complex"│
└──────────┬──────────┘
│
┌─────┴─────┐
│ │
▼ ▼
gpt-4.1-mini gpt-4.1
(rápido, (potente,
económico) preciso)
│ │
└─────┬─────┘
│
▼
Respuesta al usuario
No necesitas un modelo para clasificar: heurísticas simples basadas en texto son suficientes para la mayoría de casos.
Clasificando complejidad: heurísticas simples
Antes de implementar el routing, necesitas una función que clasifique cada request. Empecemos con heurísticas basadas en texto.
Clasificación por longitud y keywords
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
def classify_complexity(messages) -> str:
"""Clasifica la complejidad del request en 'simple' o 'complex'."""
last_message = messages[-1].content if messages else ""
complex_keywords = [
"analiza", "compara", "investiga", "explica en detalle",
"multi-step", "evalúa", "diseña", "arquitectura",
"pros y contras", "trade-offs", "profundidad"
]
is_long = len(last_message) > 200
has_complex_keywords = any(kw in last_message.lower() for kw in complex_keywords)
if is_long or has_complex_keywords:
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es un concepto clave en tecnología."
model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(model, [search])
from langchain_core.messages import HumanMessage
simple_msgs = [HumanMessage(content="¿Qué es Python?")]
complex_msgs = [HumanMessage(content="Analiza en detalle las diferencias entre Python y Rust para sistemas de backend distribuido, considerando rendimiento, seguridad de memoria y ecosistema.")]
print(f"'¿Qué es Python?' → {classify_complexity(simple_msgs)}")
print(f"'Analiza en detalle...' → {classify_complexity(complex_msgs)}")
# Output esperado:
# '¿Qué es Python?' → simple
# 'Analiza en detalle...' → complex
Clasificación multi-nivel
Para routing más granular, puedes usar tres niveles en lugar de dos:
from dotenv import load_dotenv
load_dotenv()
from langchain_core.messages import HumanMessage
def classify_complexity_v2(messages) -> str:
"""Clasifica complejidad en 'trivial', 'moderate', o 'complex'."""
last_message = messages[-1].content if messages else ""
text = last_message.lower()
trivial_patterns = ["hola", "gracias", "ok", "sí", "no", "adiós"]
complex_keywords = [
"analiza", "compara", "investiga", "explica en detalle",
"diseña", "arquitectura", "evalúa", "profundidad"
]
if any(text.strip() == p for p in trivial_patterns):
return "trivial"
if len(last_message) > 200 or any(kw in text for kw in complex_keywords):
return "complex"
return "moderate"
test_cases = [
"Hola",
"¿Qué es Python?",
"Analiza las diferencias entre microservicios y monolitos considerando escalabilidad, mantenimiento, deployment y costos operativos de cada arquitectura.",
]
for msg_text in test_cases:
msgs = [HumanMessage(content=msg_text)]
level = classify_complexity_v2(msgs)
print(f"[{level:>8}] {msg_text[:60]}...")
# Output esperado:
# [ trivial] Hola...
# [moderate] ¿Qué es Python?...
# [ complex] Analiza las diferencias entre microservicios y monolitos co...
Implementando routing con @wrap_model_call
Aquí es donde todo se conecta. Usas @wrap_model_call para interceptar la llamada al modelo y reemplazar el modelo según la clasificación de complejidad.
Routing básico: dos modelos
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
def classify_complexity(messages) -> str:
"""Clasifica la complejidad del request."""
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle", "multi-step"]
is_complex = (
len(last_message) > 200 or
any(kw in last_message.lower() for kw in complex_keywords)
)
return "complex" if is_complex else "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} — información relevante encontrada."
def wrap_model_call(messages, config, call_next):
"""Selecciona el modelo según la complejidad de la pregunta."""
complexity = classify_complexity(messages)
if complexity == "simple":
print(f"[ROUTING] → gpt-4.1-mini (simple)")
response = cheap_model.invoke(messages)
else:
print(f"[ROUTING] → gpt-4.1 (complejo)")
response = powerful_model.invoke(messages)
return response
agent = create_agent(
cheap_model,
[search],
wrap_model_call=wrap_model_call,
)
result = agent.invoke({"messages": [("user", "¿Qué es Python?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}")
# Output esperado:
# [ROUTING] → gpt-4.1-mini (simple)
# [ROUTING] → gpt-4.1-mini (simple)
#
# Respuesta: Python es un lenguaje de programación de alto nivel, interpretado y de propósito general...
Nota que [ROUTING] puede aparecer múltiples veces: una por la primera llamada al modelo (que decide si llamar tools), y otra cuando el agente genera la respuesta final.
Routing con logging de costos
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COST_PER_CALL = {
"gpt-4.1-mini": 0.0004,
"gpt-4.1": 0.005,
}
routing_log = []
def classify_complexity(messages) -> str:
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es un concepto de tecnología moderna."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
if complexity == "simple":
model_name = "gpt-4.1-mini"
response = cheap_model.invoke(messages)
else:
model_name = "gpt-4.1"
response = powerful_model.invoke(messages)
cost = COST_PER_CALL[model_name]
routing_log.append({
"model": model_name,
"complexity": complexity,
"cost": cost,
})
print(f"[ROUTING] {model_name} | complejidad={complexity} | costo=${cost}")
return response
agent = create_agent(
cheap_model,
[search],
wrap_model_call=wrap_model_call,
)
result = agent.invoke({"messages": [("user", "¿Qué es Docker?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}")
print(f"\n--- Resumen de routing ---")
total_cost = sum(entry["cost"] for entry in routing_log)
print(f"Llamadas totales: {len(routing_log)}")
print(f"Costo total: ${total_cost:.4f}")
print(f"Costo si todo fuera gpt-4.1: ${len(routing_log) * COST_PER_CALL['gpt-4.1']:.4f}")
print(f"Ahorro: ${(len(routing_log) * COST_PER_CALL['gpt-4.1']) - total_cost:.4f}")
# Output esperado:
# [ROUTING] gpt-4.1-mini | complejidad=simple | costo=$0.0004
# [ROUTING] gpt-4.1-mini | complejidad=simple | costo=$0.0004
#
# Respuesta: Docker es una plataforma de contenedores que permite empaquetar aplicaciones con todas sus...
#
# --- Resumen de routing ---
# Llamadas totales: 2
# Costo total: $0.0008
# Costo si todo fuera gpt-4.1: $0.0100
# Ahorro: $0.0092
init_chat_model para selección dinámica
init_chat_model acepta un string con formato "provider:model_name", lo que te permite inicializar modelos de diferentes proveedores con la misma interfaz.
Inicialización de múltiples proveedores
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
models = {
"cheap": init_chat_model("openai:gpt-4.1-mini"),
"balanced": init_chat_model("openai:gpt-4.1"),
"powerful": init_chat_model("anthropic:claude-sonnet-4-20250514"),
}
for name, model in models.items():
response = model.invoke([("user", "Di 'hola' en una palabra.")])
print(f"[{name:>9}] {response.content}")
# Output esperado:
# [ cheap] Hola
# [ balanced] Hola
# [ powerful] Hola
Routing con init_chat_model dinámico
En lugar de pre-inicializar todos los modelos, puedes crear el modelo on-demand:
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
MODEL_REGISTRY = {
"simple": "openai:gpt-4.1-mini",
"complex": "openai:gpt-4.1",
}
model_cache = {}
def get_model(complexity: str):
"""Obtiene o crea un modelo según la complejidad."""
if complexity not in model_cache:
model_id = MODEL_REGISTRY[complexity]
model_cache[complexity] = init_chat_model(model_id)
print(f"[CACHE] Modelo '{model_id}' inicializado y cacheado")
return model_cache[complexity]
def classify_complexity(messages) -> str:
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es relevante en el contexto actual."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
model = get_model(complexity)
model_id = MODEL_REGISTRY[complexity]
print(f"[ROUTING] → {model_id} ({complexity})")
return model.invoke(messages)
base_model = init_chat_model("openai:gpt-4.1-mini")
agent = create_agent(base_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "¿Qué es Kubernetes?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}")
# Output esperado:
# [CACHE] Modelo 'openai:gpt-4.1-mini' inicializado y cacheado
# [ROUTING] → openai:gpt-4.1-mini (simple)
# [ROUTING] → openai:gpt-4.1-mini (simple)
#
# Respuesta: Kubernetes es una plataforma de orquestación de contenedores de código abierto que autom...
El cache evita re-inicializar modelos en cada llamada — los modelos se crean una sola vez y se reutilizan.
Optimización de costos: tracking de savings
En producción, necesitas métricas reales para justificar el routing. Este patrón trackea el ahorro acumulado.
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COST_PER_1K_TOKENS = {
"gpt-4.1-mini": {"input": 0.0004, "output": 0.0016},
"gpt-4.1": {"input": 0.002, "output": 0.008},
}
class CostTracker:
def __init__(self):
self.calls = []
def log(self, model_name: str, complexity: str):
baseline_cost = COST_PER_1K_TOKENS["gpt-4.1"]["input"]
actual_cost = COST_PER_1K_TOKENS[model_name]["input"]
self.calls.append({
"model": model_name,
"complexity": complexity,
"actual_cost": actual_cost,
"baseline_cost": baseline_cost,
})
def summary(self) -> dict:
total_actual = sum(c["actual_cost"] for c in self.calls)
total_baseline = sum(c["baseline_cost"] for c in self.calls)
savings = total_baseline - total_actual
pct = (savings / total_baseline * 100) if total_baseline > 0 else 0
return {
"total_calls": len(self.calls),
"simple_calls": sum(1 for c in self.calls if c["complexity"] == "simple"),
"complex_calls": sum(1 for c in self.calls if c["complexity"] == "complex"),
"actual_cost": total_actual,
"baseline_cost": total_baseline,
"savings": savings,
"savings_pct": pct,
}
tracker = CostTracker()
def classify_complexity(messages) -> str:
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es un tema relevante."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
if complexity == "simple":
model_name = "gpt-4.1-mini"
response = cheap_model.invoke(messages)
else:
model_name = "gpt-4.1"
response = powerful_model.invoke(messages)
tracker.log(model_name, complexity)
return response
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
questions = [
"¿Qué es Python?",
"Hola",
"Analiza las diferencias entre REST y GraphQL en detalle",
"¿Cuánto es 2+2?",
"Compara los pros y contras de microservicios vs monolitos",
]
for q in questions:
print(f"\n--- Pregunta: {q[:50]}... ---")
result = agent.invoke({"messages": [("user", q)]})
print(f"Respuesta: {result['messages'][-1].content[:80]}...")
s = tracker.summary()
print(f"\n{'='*50}")
print(f"RESUMEN DE COSTOS")
print(f"{'='*50}")
print(f"Total llamadas al modelo: {s['total_calls']}")
print(f" → Simples (gpt-4.1-mini): {s['simple_calls']}")
print(f" → Complejas (gpt-4.1): {s['complex_calls']}")
print(f"Costo real: ${s['actual_cost']:.4f}/1K tokens")
print(f"Costo baseline: ${s['baseline_cost']:.4f}/1K tokens")
print(f"Ahorro: ${s['savings']:.4f} ({s['savings_pct']:.1f}%)")
# Output esperado:
# --- Pregunta: ¿Qué es Python?... ---
# Respuesta: Python es un lenguaje de programación de alto nivel, interpretado y de propósito...
#
# --- Pregunta: Hola... ---
# Respuesta: ¡Hola! ¿En qué puedo ayudarte?...
#
# --- Pregunta: Analiza las diferencias entre REST y GraphQL en de... ---
# Respuesta: REST y GraphQL son dos paradigmas diferentes para diseñar APIs. REST sigue una a...
#
# --- Pregunta: ¿Cuánto es 2+2?... ---
# Respuesta: 2 + 2 = 4...
#
# --- Pregunta: Compara los pros y contras de microservicios vs mo... ---
# Respuesta: Los microservicios y los monolitos representan dos enfoques arquitectónicos dife...
#
# ==================================================
# RESUMEN DE COSTOS
# ==================================================
# Total llamadas al modelo: 10
# → Simples (gpt-4.1-mini): 6
# → Complejas (gpt-4.1): 4
# Costo real: $0.0104/1K tokens
# Costo baseline: $0.0200/1K tokens
# Ahorro: $0.0096 (48.0%)
Optimización de latencia: modelo rápido para real-time
El routing no solo ahorra costos — también reduce latencia. Modelos más pequeños responden más rápido, lo que mejora la experiencia del usuario en interfaces interactivas.
from dotenv import load_dotenv
load_dotenv()
import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
fast_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
def classify_complexity(messages) -> str:
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado para '{query}': dato relevante encontrado."
latency_log = []
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
start = time.time()
if complexity == "simple":
model_name = "gpt-4.1-mini"
response = fast_model.invoke(messages)
else:
model_name = "gpt-4.1"
response = powerful_model.invoke(messages)
elapsed_ms = (time.time() - start) * 1000
latency_log.append({"model": model_name, "latency_ms": elapsed_ms})
print(f"[LATENCY] {model_name}: {elapsed_ms:.0f}ms")
return response
agent = create_agent(fast_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "¿Qué es Docker?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:80]}")
if latency_log:
avg_latency = sum(e["latency_ms"] for e in latency_log) / len(latency_log)
print(f"\nLatencia promedio: {avg_latency:.0f}ms en {len(latency_log)} llamadas")
# Output esperado:
# [LATENCY] gpt-4.1-mini: 450ms
# [LATENCY] gpt-4.1-mini: 380ms
#
# Respuesta: Docker es una plataforma de contenedores que permite empaquetar aplicaciones co
#
# Latencia promedio: 415ms en 2 llamadas
Multi-criteria routing: complejidad + presupuesto + latencia
En sistemas reales, la decisión de routing no depende solo de la complejidad. Puedes considerar múltiples factores: presupuesto restante, requisitos de latencia, y el tipo de tarea.
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from dataclasses import dataclass
@dataclass
class RoutingConfig:
budget_remaining: float = 1.0
max_latency_ms: float = 5000
prefer_quality: bool = False
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
MODEL_COSTS = {
"gpt-4.1-mini": 0.0004,
"gpt-4.1": 0.005,
}
routing_config = RoutingConfig(budget_remaining=0.05, max_latency_ms=3000)
def classify_complexity(messages) -> str:
last_message = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_message) > 200 or any(kw in last_message.lower() for kw in complex_keywords):
return "complex"
return "simple"
def select_model(complexity: str, config: RoutingConfig) -> tuple:
"""Selecciona modelo considerando múltiples criterios."""
if config.budget_remaining < MODEL_COSTS["gpt-4.1"]:
return cheap_model, "gpt-4.1-mini", "presupuesto insuficiente para gpt-4.1"
if config.max_latency_ms < 1000:
return cheap_model, "gpt-4.1-mini", "latencia requerida < 1s"
if complexity == "complex" or config.prefer_quality:
return powerful_model, "gpt-4.1", "complejidad alta o preferencia de calidad"
return cheap_model, "gpt-4.1-mini", "request simple, optimizando costo"
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es un concepto relevante en ingeniería."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
model, model_name, reason = select_model(complexity, routing_config)
cost = MODEL_COSTS[model_name]
routing_config.budget_remaining -= cost
print(f"[ROUTING] {model_name} | razón: {reason}")
print(f"[BUDGET] restante: ${routing_config.budget_remaining:.4f}")
return model.invoke(messages)
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({
"messages": [("user", "¿Qué es Kubernetes?")]
})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}")
# Output esperado:
# [ROUTING] gpt-4.1-mini | razón: request simple, optimizando costo
# [BUDGET] restante: $0.0496
# [ROUTING] gpt-4.1-mini | razón: request simple, optimizando costo
# [BUDGET] restante: $0.0492
#
# Respuesta: Kubernetes es una plataforma de orquestación de contenedores de código abierto que facilita el des...
Fallback patterns: escalamiento por calidad
Un patrón avanzado: enviar primero al modelo económico, evaluar la calidad de la respuesta, y si no es suficiente, re-enviar al modelo potente. Esto combina lo mejor de ambos mundos.
Fallback simple: por longitud de respuesta
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
MIN_RESPONSE_LENGTH = 50
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} tiene múltiples aplicaciones prácticas."
def wrap_model_call(messages, config, call_next):
print(f"[FALLBACK] Intentando con gpt-4.1-mini...")
response = cheap_model.invoke(messages)
if len(response.content) < MIN_RESPONSE_LENGTH:
print(f"[FALLBACK] Respuesta corta ({len(response.content)} chars), escalando a gpt-4.1...")
response = powerful_model.invoke(messages)
print(f"[FALLBACK] gpt-4.1 respondió ({len(response.content)} chars)")
else:
print(f"[FALLBACK] gpt-4.1-mini suficiente ({len(response.content)} chars)")
return response
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "¿Qué es LangChain?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:120]}")
# Output esperado:
# [FALLBACK] Intentando con gpt-4.1-mini...
# [FALLBACK] gpt-4.1-mini suficiente (187 chars)
# [FALLBACK] Intentando con gpt-4.1-mini...
# [FALLBACK] gpt-4.1-mini suficiente (203 chars)
#
# Respuesta: LangChain es un framework de código abierto diseñado para facilitar la creación de aplicaciones que utilizan modelos de...
Fallback con verificación de contenido
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
def quality_check(response_text: str) -> bool:
"""Verifica si la respuesta cumple criterios mínimos de calidad."""
if len(response_text) < 30:
return False
low_quality_indicators = [
"no sé", "no tengo información", "no puedo",
"como modelo de lenguaje", "i don't know"
]
if any(indicator in response_text.lower() for indicator in low_quality_indicators):
return False
return True
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Resultado: {query} es un concepto técnico importante."
fallback_stats = {"direct": 0, "escalated": 0}
def wrap_model_call(messages, config, call_next):
response = cheap_model.invoke(messages)
if quality_check(response.content):
fallback_stats["direct"] += 1
print(f"[QUALITY] ✅ gpt-4.1-mini pasó quality check")
return response
fallback_stats["escalated"] += 1
print(f"[QUALITY] ⚠️ Escalando a gpt-4.1")
return powerful_model.invoke(messages)
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "Explica qué es Docker")]})
print(f"\nRespuesta: {result['messages'][-1].content[:100]}")
print(f"\nEstadísticas: {fallback_stats}")
# Output esperado:
# [QUALITY] ✅ gpt-4.1-mini pasó quality check
# [QUALITY] ✅ gpt-4.1-mini pasó quality check
#
# Respuesta: Docker es una plataforma de contenedorización que permite empaquetar aplicaciones junto con todas...
#
# Estadísticas: {'direct': 2, 'escalated': 0}
Comparación: static model vs dynamic routing
¿Cuándo vale la pena implementar dynamic routing vs usar un solo modelo?
| Criterio | Modelo estático | Dynamic routing |
|---|---|---|
| Complejidad de implementación | Ninguna | Media |
| Costo por request | Fijo (alto si usas modelo potente) | Variable (bajo promedio) |
| Calidad de respuesta | Consistente | Variable (alta donde importa) |
| Latencia | Fija | Variable (baja para requests simples) |
| Debugging | Simple | Requiere logging del routing |
| Mantenimiento | Bajo | Medio (ajustar heurísticas) |
Cuándo usar cada enfoque
- ✅ Modelo estático — Prototipos, aplicaciones internas, presupuesto no es preocupación
- ✅ Dynamic routing — Producción con volumen alto, mezcla de requests simples/complejos
- ⚠️ No uses dynamic routing si tu volumen es bajo (<100 requests/día) — la complejidad no se justifica
- ❌ Nunca uses solo el modelo económico para todo si la calidad importa en preguntas complejas
Ejemplo comparativo
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COST_PER_CALL = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}
request_profile = {
"simple": 70,
"complex": 30,
}
cost_static_powerful = 100 * COST_PER_CALL["gpt-4.1"]
cost_static_cheap = 100 * COST_PER_CALL["gpt-4.1-mini"]
cost_dynamic = (
request_profile["simple"] * COST_PER_CALL["gpt-4.1-mini"] +
request_profile["complex"] * COST_PER_CALL["gpt-4.1"]
)
print("Escenario: 100 requests (70% simples, 30% complejos)")
print(f" Solo gpt-4.1: ${cost_static_powerful:.4f}")
print(f" Solo gpt-4.1-mini: ${cost_static_cheap:.4f} (calidad baja en complejos)")
print(f" Dynamic routing: ${cost_dynamic:.4f} (calidad alta donde importa)")
print(f"\nAhorro vs gpt-4.1 estático: {((cost_static_powerful - cost_dynamic) / cost_static_powerful * 100):.1f}%")
# Output esperado:
# Escenario: 100 requests (70% simples, 30% complejos)
# Solo gpt-4.1: $0.5000
# Solo gpt-4.1-mini: $0.0400 (calidad baja en complejos)
# Dynamic routing: $0.1780 (calidad alta donde importa)
#
# Ahorro vs gpt-4.1 estático: 64.4%
Conexión con el proyecto
En el Proyecto del módulo (Cápsula 08), implementarás un agente completo con dynamic model routing: el agente selecciona automáticamente entre modelo económico y potente según la complejidad, combinado con logging middleware para monitorear cada decisión de routing. Esto se integrará con dynamic tools (Cápsula 06) y AgentMiddleware class (Cápsula 07) para crear un sistema de middleware compuesto.
Troubleshooting
Problema 1: El modelo siempre se clasifica como "simple"
Causa: Tus keywords de complejidad no coinciden con los mensajes reales del usuario, o estás accediendo al mensaje equivocado.
Solución: Verifica qué mensaje estás analizando — en el loop del agente, messages incluye el historial completo:
def classify_complexity(messages):
last_message = messages[-1].content if messages else ""
print(f"[DEBUG] Analizando: '{last_message[:80]}'")
# ... rest of classification
Problema 2: wrap_model_call no intercepta todas las llamadas
Causa: Algunas llamadas internas del agente pueden no pasar por el wrapper si no están configuradas correctamente.
Solución: Asegúrate de pasar wrap_model_call como parámetro de create_agent:
agent = create_agent(
model,
tools,
wrap_model_call=wrap_model_call, # No como decorador, sino como parámetro
)
Problema 3: El cache de modelos crece sin límite
Causa: Si usas init_chat_model dentro del wrapper sin cache, creas un modelo nuevo en cada llamada.
Solución: Pre-inicializa los modelos fuera del wrapper o usa un diccionario de cache:
model_cache = {}
def get_model(model_id):
if model_id not in model_cache:
model_cache[model_id] = init_chat_model(model_id)
return model_cache[model_id]
Problema 4: El fallback siempre se activa (respuestas siempre "cortas")
Causa: Tu umbral de calidad es demasiado alto o estás midiendo antes de que el modelo complete la respuesta.
Solución: Ajusta los umbrales y verifica que response.content contiene la respuesta completa:
print(f"[DEBUG] Longitud respuesta: {len(response.content)}")
print(f"[DEBUG] Contenido: {response.content[:200]}")
Problema 5: Costos no coinciden con lo esperado
Causa: Cada invocación del agente puede hacer múltiples llamadas al modelo (una para decidir tools, otra para la respuesta final), y el wrapper se ejecuta en cada una.
Solución: Ten en cuenta que un agent.invoke() puede generar 2+ llamadas al wrapper:
def wrap_model_call(messages, config, call_next):
# Esta función se ejecuta N veces por invoke(), no solo una
print(f"[DEBUG] Llamada #{len(routing_log) + 1}")
# ...
Ejercicios
Ejercicio 1: Routing básico con dos modelos (Fácil)
Crea un agente con wrap_model_call que envíe preguntas con menos de 50 caracteres a gpt-4.1-mini y preguntas más largas a gpt-4.1. Imprime qué modelo se usa en cada llamada.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
@tool
def search(query: str) -> str:
"""Busca información sobre un tema."""
return f"Info: {query} es un tema de interés."
def wrap_model_call(messages, config, call_next):
last_msg = messages[-1].content if messages else ""
if len(last_msg) < 50:
print(f"[SHORT] → gpt-4.1-mini ({len(last_msg)} chars)")
return cheap_model.invoke(messages)
else:
print(f"[LONG] → gpt-4.1 ({len(last_msg)} chars)")
return powerful_model.invoke(messages)
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "¿Qué es Python?")]})
print(f"Respuesta: {result['messages'][-1].content[:80]}")
# Output esperado:
# [SHORT] → gpt-4.1-mini (15 chars)
# [SHORT] → gpt-4.1-mini (15 chars)
# Respuesta: Python es un lenguaje de programación de alto nivel, interpretado y de propósito...
Explicación: El routing más simple posible: la longitud del mensaje determina el modelo. En producción, usarías heurísticas más sofisticadas, pero el principio es el mismo.
Ejercicio 2: CostTracker con resumen (Fácil)
Crea una clase CostTracker que registre cada llamada al modelo con su costo estimado. Al final, imprime un resumen con el total gastado y el ahorro vs usar siempre el modelo caro.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}
class CostTracker:
def __init__(self):
self.entries = []
def log(self, model_name: str):
self.entries.append({"model": model_name, "cost": COSTS[model_name]})
def summary(self):
total = sum(e["cost"] for e in self.entries)
baseline = len(self.entries) * COSTS["gpt-4.1"]
saving = baseline - total
print(f"Llamadas: {len(self.entries)}")
print(f"Costo real: ${total:.4f}")
print(f"Costo baseline: ${baseline:.4f}")
print(f"Ahorro: ${saving:.4f} ({saving/baseline*100:.1f}%)")
tracker = CostTracker()
def classify_complexity(messages) -> str:
last_msg = messages[-1].content if messages else ""
if len(last_msg) > 100 or "analiza" in last_msg.lower():
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información."""
return f"Resultado: {query}."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
if complexity == "simple":
tracker.log("gpt-4.1-mini")
return cheap_model.invoke(messages)
else:
tracker.log("gpt-4.1")
return powerful_model.invoke(messages)
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
for q in ["Hola", "¿Qué es Python?", "Analiza REST vs GraphQL"]:
agent.invoke({"messages": [("user", q)]})
print("\n--- Resumen ---")
tracker.summary()
# Output esperado:
# --- Resumen ---
# Llamadas: 6
# Costo real: $0.0116
# Costo baseline: $0.0300
# Ahorro: $0.0184 (61.3%)
Explicación: CostTracker acumula el costo de cada llamada. El resumen compara contra el costo de usar siempre el modelo potente, mostrando el ahorro del routing.
Ejercicio 3: Clasificación multi-nivel (Medio)
Implementa clasificación de tres niveles (trivial, moderate, complex) y haz que cada nivel use un modelo diferente: gpt-4.1-mini para trivial, gpt-4.1-mini con temperatura alta para moderate, y gpt-4.1 para complex.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
trivial_model = init_chat_model("openai:gpt-4.1-mini", temperature=0)
moderate_model = init_chat_model("openai:gpt-4.1-mini", temperature=0.7)
complex_model = init_chat_model("openai:gpt-4.1")
def classify_three_levels(messages) -> str:
last_msg = messages[-1].content if messages else ""
text = last_msg.lower().strip()
trivial_patterns = ["hola", "gracias", "ok", "sí", "no", "adiós", "bye"]
complex_keywords = ["analiza", "compara", "investiga", "diseña", "explica en detalle"]
if text in trivial_patterns or len(text) < 10:
return "trivial"
if len(last_msg) > 200 or any(kw in text for kw in complex_keywords):
return "complex"
return "moderate"
@tool
def search(query: str) -> str:
"""Busca información."""
return f"Resultado: {query} es relevante."
routing_history = []
def wrap_model_call(messages, config, call_next):
level = classify_three_levels(messages)
models = {
"trivial": (trivial_model, "gpt-4.1-mini (t=0)"),
"moderate": (moderate_model, "gpt-4.1-mini (t=0.7)"),
"complex": (complex_model, "gpt-4.1"),
}
model, name = models[level]
routing_history.append({"level": level, "model": name})
print(f"[{level:>8}] → {name}")
return model.invoke(messages)
agent = create_agent(trivial_model, [search], wrap_model_call=wrap_model_call)
for q in ["Hola", "¿Qué es Python?", "Analiza microservicios vs monolitos"]:
print(f"\nPregunta: {q}")
result = agent.invoke({"messages": [("user", q)]})
print(f"Respuesta: {result['messages'][-1].content[:60]}...")
print(f"\n--- Historial de routing ---")
for entry in routing_history:
print(f" [{entry['level']:>8}] {entry['model']}")
# Output esperado:
# Pregunta: Hola
# [ trivial] → gpt-4.1-mini (t=0)
# Respuesta: ¡Hola! ¿En qué puedo ayudarte?...
#
# Pregunta: ¿Qué es Python?
# [moderate] → gpt-4.1-mini (t=0.7)
# [moderate] → gpt-4.1-mini (t=0.7)
# Respuesta: Python es un lenguaje de programación de alto nivel, interpr...
#
# Pregunta: Analiza microservicios vs monolitos
# [ complex] → gpt-4.1
# [ complex] → gpt-4.1
# Respuesta: Los microservicios y los monolitos son dos enfoques arquite...
#
# --- Historial de routing ---
# [ trivial] gpt-4.1-mini (t=0)
# [moderate] gpt-4.1-mini (t=0.7)
# [moderate] gpt-4.1-mini (t=0.7)
# [ complex] gpt-4.1
# [ complex] gpt-4.1
Explicación: Tres niveles de clasificación permiten mayor granularidad. Para moderate, usamos el mismo modelo económico pero con temperatura más alta para dar respuestas más creativas, sin el costo de gpt-4.1.
Ejercicio 4: Fallback con quality check (Medio)
Implementa un fallback que use gpt-4.1-mini primero. Si la respuesta contiene frases como "no sé" o "no tengo información", escala automáticamente a gpt-4.1.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
LOW_QUALITY_PHRASES = [
"no sé", "no tengo información", "no puedo responder",
"como modelo de lenguaje", "no tengo acceso",
]
def is_low_quality(text: str) -> bool:
return any(phrase in text.lower() for phrase in LOW_QUALITY_PHRASES)
@tool
def search(query: str) -> str:
"""Busca información."""
return f"Resultado: {query} tiene aplicaciones en producción."
fallback_log = {"accepted": 0, "escalated": 0}
def wrap_model_call(messages, config, call_next):
response = cheap_model.invoke(messages)
if is_low_quality(response.content):
fallback_log["escalated"] += 1
print(f"[FALLBACK] ⚠️ Calidad baja detectada, escalando a gpt-4.1")
return powerful_model.invoke(messages)
fallback_log["accepted"] += 1
print(f"[FALLBACK] ✅ gpt-4.1-mini aceptado")
return response
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
result = agent.invoke({"messages": [("user", "¿Qué es FastAPI?")]})
print(f"\nRespuesta: {result['messages'][-1].content[:80]}")
print(f"\nEstadísticas: {fallback_log}")
# Output esperado:
# [FALLBACK] ✅ gpt-4.1-mini aceptado
# [FALLBACK] ✅ gpt-4.1-mini aceptado
#
# Respuesta: FastAPI es un framework moderno y rápido para construir APIs con Python, basado en...
#
# Estadísticas: {'accepted': 2, 'escalated': 0}
Explicación: El fallback evalúa la respuesta del modelo económico. Si detecta frases que indican incapacidad de responder, escala al modelo potente. Esto asegura calidad sin pagar siempre el precio alto.
Ejercicio 5: Routing con presupuesto limitado (Difícil)
Crea un sistema de routing que respete un presupuesto máximo. Si el presupuesto está por agotarse, fuerza el modelo económico sin importar la complejidad. Simula 5 preguntas y muestra el presupuesto restante después de cada una.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}
class BudgetRouter:
def __init__(self, budget: float):
self.budget = budget
self.initial_budget = budget
self.history = []
def select_and_charge(self, complexity: str) -> tuple:
if self.budget < COSTS["gpt-4.1"]:
model_name = "gpt-4.1-mini"
reason = "presupuesto forzado"
elif complexity == "complex":
model_name = "gpt-4.1"
reason = "complejidad alta"
else:
model_name = "gpt-4.1-mini"
reason = "request simple"
cost = COSTS[model_name]
self.budget -= cost
self.history.append({
"model": model_name, "cost": cost,
"budget_after": self.budget, "reason": reason,
})
return (powerful_model if model_name == "gpt-4.1" else cheap_model), model_name, reason
router = BudgetRouter(budget=0.015)
def classify_complexity(messages) -> str:
last_msg = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga"]
if any(kw in last_msg.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información."""
return f"Resultado: {query}."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
model, name, reason = router.select_and_charge(complexity)
print(f" [{name}] razón={reason} | restante=${router.budget:.4f}")
return model.invoke(messages)
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
questions = [
"¿Qué es Python?",
"Analiza pros y contras de Docker",
"Hola",
"Compara Kubernetes vs Docker Swarm",
"¿Qué hora es?",
]
for i, q in enumerate(questions, 1):
print(f"\n[Q{i}] {q}")
result = agent.invoke({"messages": [("user", q)]})
print(f" Respuesta: {result['messages'][-1].content[:60]}...")
print(f"\n{'='*50}")
print(f"Presupuesto inicial: ${router.initial_budget:.4f}")
print(f"Presupuesto final: ${router.budget:.4f}")
print(f"Gastado: ${router.initial_budget - router.budget:.4f}")
# Output esperado:
# [Q1] ¿Qué es Python?
# [gpt-4.1-mini] razón=request simple | restante=$0.0146
# [gpt-4.1-mini] razón=request simple | restante=$0.0142
# Respuesta: Python es un lenguaje de programación de alto nivel, interp...
#
# [Q2] Analiza pros y contras de Docker
# [gpt-4.1] razón=complejidad alta | restante=$0.0092
# [gpt-4.1] razón=complejidad alta | restante=$0.0042
# Respuesta: Docker presenta varios pros y contras que vale la pena con...
#
# [Q3] Hola
# [gpt-4.1-mini] razón=request simple | restante=$0.0038
# Respuesta: ¡Hola! ¿En qué puedo ayudarte?...
#
# [Q4] Compara Kubernetes vs Docker Swarm
# [gpt-4.1-mini] razón=presupuesto forzado | restante=$0.0034
# [gpt-4.1-mini] razón=presupuesto forzado | restante=$0.0030
# Respuesta: Kubernetes y Docker Swarm son plataformas de orquestación...
#
# [Q5] ¿Qué hora es?
# [gpt-4.1-mini] razón=request simple | restante=$0.0026
# Respuesta: No tengo acceso a la hora actual, pero puedes verificarla...
#
# ==================================================
# Presupuesto inicial: $0.0150
# Presupuesto final: $0.0026
# Gastado: $0.0124
Explicación: BudgetRouter gestiona un presupuesto finito. Cuando queda menos de lo que cuesta una llamada al modelo potente, fuerza el modelo económico. Nota cómo Q4 ("Compara...") normalmente iría a gpt-4.1, pero el presupuesto insuficiente fuerza gpt-4.1-mini.
Ejercicio 6: Dashboard de routing en tiempo real (Difícil)
Crea un sistema completo que combine routing por complejidad, cost tracking, y latency tracking. Al final, imprime un dashboard con todas las métricas.
Ver solución
from dotenv import load_dotenv
load_dotenv()
import time
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
cheap_model = init_chat_model("openai:gpt-4.1-mini")
powerful_model = init_chat_model("openai:gpt-4.1")
COSTS = {"gpt-4.1-mini": 0.0004, "gpt-4.1": 0.005}
class RoutingDashboard:
def __init__(self):
self.entries = []
def log(self, model: str, complexity: str, latency_ms: float):
self.entries.append({
"model": model, "complexity": complexity,
"latency_ms": latency_ms, "cost": COSTS[model],
})
def print_dashboard(self):
if not self.entries:
print("No hay datos.")
return
total_cost = sum(e["cost"] for e in self.entries)
baseline_cost = len(self.entries) * COSTS["gpt-4.1"]
avg_latency = sum(e["latency_ms"] for e in self.entries) / len(self.entries)
simple_count = sum(1 for e in self.entries if e["complexity"] == "simple")
complex_count = sum(1 for e in self.entries if e["complexity"] == "complex")
simple_lat = [e["latency_ms"] for e in self.entries if e["complexity"] == "simple"]
complex_lat = [e["latency_ms"] for e in self.entries if e["complexity"] == "complex"]
print(f"\n{'='*55}")
print(f" ROUTING DASHBOARD")
print(f"{'='*55}")
print(f" Total llamadas: {len(self.entries)}")
print(f" Simples: {simple_count}")
print(f" Complejas: {complex_count}")
print(f"{'─'*55}")
print(f" Costo total: ${total_cost:.4f}")
print(f" Costo baseline: ${baseline_cost:.4f}")
print(f" Ahorro: ${baseline_cost - total_cost:.4f} ({(baseline_cost - total_cost)/baseline_cost*100:.1f}%)")
print(f"{'─'*55}")
print(f" Latencia promedio: {avg_latency:.0f}ms")
if simple_lat:
print(f" Latencia simples: {sum(simple_lat)/len(simple_lat):.0f}ms")
if complex_lat:
print(f" Latencia complejas: {sum(complex_lat)/len(complex_lat):.0f}ms")
print(f"{'='*55}")
dashboard = RoutingDashboard()
def classify_complexity(messages) -> str:
last_msg = messages[-1].content if messages else ""
complex_keywords = ["analiza", "compara", "investiga", "explica en detalle"]
if len(last_msg) > 150 or any(kw in last_msg.lower() for kw in complex_keywords):
return "complex"
return "simple"
@tool
def search(query: str) -> str:
"""Busca información."""
return f"Resultado: {query} es un concepto importante."
def wrap_model_call(messages, config, call_next):
complexity = classify_complexity(messages)
start = time.time()
if complexity == "simple":
model_name = "gpt-4.1-mini"
response = cheap_model.invoke(messages)
else:
model_name = "gpt-4.1"
response = powerful_model.invoke(messages)
latency_ms = (time.time() - start) * 1000
dashboard.log(model_name, complexity, latency_ms)
return response
agent = create_agent(cheap_model, [search], wrap_model_call=wrap_model_call)
questions = [
"¿Qué es Docker?",
"Analiza en detalle las ventajas de microservicios sobre monolitos",
"Hola",
"¿Qué es Python?",
"Compara React, Vue y Angular para aplicaciones enterprise",
]
for q in questions:
print(f"→ {q[:50]}...")
agent.invoke({"messages": [("user", q)]})
dashboard.print_dashboard()
# Output esperado:
# → ¿Qué es Docker?...
# → Analiza en detalle las ventajas de microservicios...
# → Hola...
# → ¿Qué es Python?...
# → Compara React, Vue y Angular para aplicaciones en...
#
# =======================================================
# ROUTING DASHBOARD
# =======================================================
# Total llamadas: 10
# Simples: 6
# Complejas: 4
# ───────────────────────────────────────────────────────
# Costo total: $0.0224
# Costo baseline: $0.0500
# Ahorro: $0.0276 (55.2%)
# ───────────────────────────────────────────────────────
# Latencia promedio: 520ms
# Latencia simples: 380ms
# Latencia complejas: 730ms
# =======================================================
Explicación: El RoutingDashboard combina cost tracking y latency tracking en un solo reporte. Esto te da visibilidad completa sobre el impacto del routing dinámico en tu sistema.
Resumen
En esta cápsula aprendiste:
- Model routing selecciona automáticamente el modelo óptimo para cada request según su complejidad
classify_complexity()usa heurísticas simples (longitud, keywords) para clasificar requests sin necesitar un modelo adicional@wrap_model_calles el punto de intercepción perfecto para implementar routing — ya sabes interceptar la llamada, ahora cambias el modelo mismoinit_chat_modelpermite inicializar modelos de múltiples proveedores ("openai:gpt-4.1-mini","anthropic:claude-sonnet") con la misma interfaz- Cost tracking mide el ahorro real del routing vs un modelo estático — típicamente 40-80% de ahorro
- Latency tracking confirma que modelos económicos son significativamente más rápidos
- Multi-criteria routing considera complejidad + presupuesto + latencia para decisiones óptimas
- Fallback patterns intentan el modelo económico primero y escalan si la calidad es insuficiente
Próxima cápsula: Dynamic Tools y Dynamic Prompts — cómo filtrar las tools disponibles por permisos de usuario y generar prompts dinámicos basados en contexto.
Recursos adicionales
- init_chat_model API Reference — Referencia de init_chat_model y proveedores soportados
- How to add custom middleware to agents — Guía oficial de middleware en agentes
- Model Routing Patterns — Patrones de routing de modelos en LangChain
- OpenAI Pricing — Precios actualizados de modelos OpenAI
- Anthropic Pricing — Precios de modelos Claude
- LangSmith Cost Tracking — Monitoreo de costos en producción con LangSmith
Módulo 4 — LangChain & LangGraph: From Chains to Agents