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 preguntaEjemploModelo 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?

CriterioModelo estáticoDynamic routing
Complejidad de implementaciónNingunaMedia
Costo por requestFijo (alto si usas modelo potente)Variable (bajo promedio)
Calidad de respuestaConsistenteVariable (alta donde importa)
LatenciaFijaVariable (baja para requests simples)
DebuggingSimpleRequiere logging del routing
MantenimientoBajoMedio (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_call es el punto de intercepción perfecto para implementar routing — ya sabes interceptar la llamada, ahora cambias el modelo mismo
  • init_chat_model permite 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

  1. init_chat_model API Reference — Referencia de init_chat_model y proveedores soportados
  2. How to add custom middleware to agents — Guía oficial de middleware en agentes
  3. Model Routing Patterns — Patrones de routing de modelos en LangChain
  4. OpenAI Pricing — Precios actualizados de modelos OpenAI
  5. Anthropic Pricing — Precios de modelos Claude
  6. LangSmith Cost Tracking — Monitoreo de costos en producción con LangSmith

Módulo 4 — LangChain & LangGraph: From Chains to Agents