Proyecto: Agente con Routing Dinámico de Modelos

Descripción del proyecto

En las siete cápsulas anteriores aprendiste a personalizar agentes sin reescribirlos: hooks individuales con @before_model y @after_model, interceptores potentes con @wrap_model_call y @wrap_tool_call, dynamic models para seleccionar el modelo en runtime, dynamic tools y dynamic prompts para adaptar el agente al contexto, y AgentMiddleware class para empaquetar todo en módulos reutilizables. Cada concepto lo viste de forma individual. Ahora vas a combinar todo en un sistema real: un agente que selecciona automáticamente el modelo según la complejidad de la pregunta.

La idea central es simple pero poderosa: no todas las preguntas requieren el mismo modelo. "¿Cuál es la capital de Francia?" no necesita GPT-4.1 — un modelo económico como GPT-4.1-mini responde perfecto y cuesta una fracción. Pero "Analiza las implicaciones de la arquitectura de microservicios vs monolito para una startup con 10 ingenieros" sí necesita capacidad de razonamiento profundo. Un agente inteligente debería hacer esta distinción automáticamente.

En este proyecto construirás exactamente eso: un middleware de clasificación que analiza la complejidad de cada pregunta, un sistema de routing que selecciona el modelo apropiado, un middleware de logging que registra cada decisión, y dynamic tools que habilitan herramientas avanzadas solo cuando la pregunta lo requiere. El resultado es un agente que optimiza costo y calidad simultáneamente — exactamente lo que necesitas en producción.

El sistema funciona como un chat interactivo en terminal donde puedes hacer preguntas simples y complejas, y ver en tiempo real qué modelo se eligió, por qué, y qué tools están disponibles.

---

Objetivo del proyecto

Construir un agente con routing dinámico de modelos que selecciona automáticamente entre un modelo económico y uno potente según la complejidad de la pregunta, con logging completo y tools dinámicos.

Al completar este proyecto:

• 🔧 Sabrás implementar un clasificador de complejidad como middleware
• 🔧 Construirás un sistema de routing que selecciona modelos en runtime
• 🔧 Agregarás logging middleware que documenta cada decisión
• 🔧 Implementarás dynamic tools por nivel de complejidad
• 🔧 Compondrás múltiples middleware en un agente production-ready
• 🔧 Tendrás un sistema funcional que integra todo el módulo

---

Especificaciones técnicas

Stack tecnológico

Componente	Versión	Propósito
Python	3.11+	Runtime
LangChain	v1.2+	Framework de LLMs + middleware
LangGraph	v1.0+	create_agent
langchain-openai	latest	Proveedor de modelos
python-dotenv	latest	Variables de entorno

Setup inicial

pip install langchain langgraph langchain-openai python-dotenv

Crea un archivo .env en la raíz de tu proyecto:

# .env
OPENAI_API_KEY=sk-...

Estructura del proyecto

agente-routing-dinamico/
├── .env                          # API key
├── dynamic_routing_agent.py      # Código principal (todo en un archivo)
└── requirements.txt              # Dependencias

# requirements.txt
langchain>=0.3.0
langgraph>=0.3.0
langchain-openai>=0.3.0
python-dotenv>=1.0.0

---

Paso 1: Definir las herramientas

El agente necesita dos niveles de herramientas: básicas (disponibles siempre) y avanzadas (solo para preguntas complejas).

Tools básicas: search y calculator

Disponibles para todo tipo de preguntas. Son operaciones rápidas y de bajo costo.

from langchain_core.tools import tool

KNOWLEDGE_BASE = {
    "python": {
        "title": "Python Programming Language",
        "content": (
            "Python es un lenguaje de alto nivel, interpretado, con tipado dinámico. "
            "Creado por Guido van Rossum en 1991. Usado en AI/ML, data science, web, "
            "automatización. Ecosistema: PyTorch, TensorFlow, Django, FastAPI. "
            "Community: >8M developers. Versión actual: 3.12."
        ),
        "source": "https://python.org",
    },
    "rust": {
        "title": "Rust Programming Language",
        "content": (
            "Rust es un lenguaje de sistemas con memory safety sin garbage collector. "
            "Creado por Mozilla en 2010. Ownership system único. Performance comparable a C/C++. "
            "Usado en: sistemas operativos, WebAssembly, CLIs, networking. "
            "Adoptado por Linux kernel, Android, Windows."
        ),
        "source": "https://rust-lang.org",
    },
    "microservicios": {
        "title": "Arquitectura de Microservicios",
        "content": (
            "Microservicios es un estilo arquitectónico que estructura una aplicación como "
            "colección de servicios pequeños, autónomos y desplegables independientemente. "
            "Ventajas: escalabilidad independiente, deploy independiente, tecnología heterogénea. "
            "Desventajas: complejidad de red, consistencia eventual, overhead operacional. "
            "Alternativa: monolito modular para equipos < 20 personas."
        ),
        "source": "https://microservices.io",
    },
    "langchain": {
        "title": "LangChain Framework",
        "content": (
            "LangChain es el framework open-source más adoptado para aplicaciones con LLMs. "
            "v1.2+ incluye: init_chat_model, create_agent, middleware system. "
            "Ecosistema: LangGraph (orquestación), LangSmith (observabilidad), Deep Agents. "
            "Usado por: startups AI, enterprise, research labs."
        ),
        "source": "https://langchain.com",
    },
}


@tool
def search(query: str) -> str:
    """Busca información general sobre un tema. Útil para preguntas factuales."""
    query_lower = query.lower()
    results = []
    for key, data in KNOWLEDGE_BASE.items():
        if key in query_lower or any(word in data["content"].lower() for word in query_lower.split()):
            results.append(f"📄 {data['title']}: {data['content']} (Fuente: {data['source']})")

    if results:
        return f"Encontré {len(results)} resultado(s):\n" + "\n\n".join(results)
    return f"Sin resultados específicos para '{query}'. Temas disponibles: {', '.join(KNOWLEDGE_BASE.keys())}"


@tool
def calculator(expression: str) -> str:
    """Evalúa expresiones matemáticas. Útil para cálculos numéricos."""
    try:
        result = eval(expression)
        return f"Resultado: {expression} = {result}"
    except Exception as e:
        return f"Error evaluando '{expression}': {e}"

Tools avanzadas: deep_research y data_analysis

Solo disponibles para preguntas complejas. Simulan operaciones más costosas y profundas.

@tool
def deep_research(topic: str, aspects: list[str]) -> str:
    """Investigación profunda sobre un tema con múltiples aspectos.
    Solo se usa para preguntas complejas que requieren análisis detallado."""
    results = []
    for aspect in aspects:
        topic_lower = topic.lower()
        if topic_lower in KNOWLEDGE_BASE:
            data = KNOWLEDGE_BASE[topic_lower]
            results.append(f"🔬 {aspect}: Basado en {data['title']} — {data['content']}")
        else:
            results.append(f"🔬 {aspect}: Análisis general sobre {topic} (sin datos específicos)")

    return (
        f"Investigación profunda sobre '{topic}':\n"
        f"Aspectos analizados: {len(aspects)}\n\n"
        + "\n\n".join(results)
    )


@tool
def data_analysis(data_description: str, analysis_type: str) -> str:
    """Analiza datos con un tipo de análisis específico.
    Solo se usa para preguntas complejas que requieren procesamiento de datos."""
    analyses = {
        "comparison": f"📊 Comparación: {data_description} — Se identificaron 3 dimensiones de comparación: rendimiento, costo, y facilidad de adopción.",
        "tradeoff": f"📊 Trade-offs: {data_description} — Ventajas vs desventajas analizadas en 4 categorías: técnica, organizacional, financiera, y temporal.",
        "recommendation": f"📊 Recomendación: {data_description} — Basado en el análisis, la recomendación depende del contexto: equipo, presupuesto, y timeline.",
        "trend": f"📊 Tendencia: {data_description} — Análisis de tendencia muestra crecimiento sostenido en los últimos 3 años.",
    }
    return analyses.get(
        analysis_type.lower(),
        f"📊 Análisis ({analysis_type}): {data_description} — Análisis completado con resultados generales."
    )

---

Paso 2: Crear el middleware clasificador de complejidad

Este middleware analiza la pregunta del usuario y la clasifica como "simple" o "complex". La clasificación se basa en indicadores heurísticos: longitud, presencia de palabras clave analíticas, y estructura de la pregunta.

import re

COMPLEXITY_KEYWORDS = [
    "analiza", "compara", "evalúa", "explica en detalle",
    "implicaciones", "trade-off", "tradeoff", "ventajas y desventajas",
    "arquitectura", "diseño", "estrategia", "profundidad",
    "vs", "versus", "diferencias entre", "pros y contras",
    "recomendación", "recomienda", "cuándo usar", "mejor opción",
    "impacto", "consecuencias", "a largo plazo",
]


def classify_complexity(text: str) -> dict:
    """Clasifica la complejidad de una pregunta."""
    text_lower = text.lower()
    word_count = len(text.split())

    keyword_matches = [kw for kw in COMPLEXITY_KEYWORDS if kw in text_lower]
    has_multiple_questions = text.count("?") > 1
    has_connectors = any(c in text_lower for c in [" y ", " pero ", " sin embargo", " además"])

    score = 0
    reasons = []

    if word_count > 20:
        score += 1
        reasons.append(f"pregunta larga ({word_count} palabras)")
    if keyword_matches:
        score += len(keyword_matches)
        reasons.append(f"keywords analíticas: {', '.join(keyword_matches[:3])}")
    if has_multiple_questions:
        score += 1
        reasons.append("múltiples preguntas")
    if has_connectors:
        score += 1
        reasons.append("conectores complejos")

    complexity = "complex" if score >= 2 else "simple"

    return {
        "complexity": complexity,
        "score": score,
        "reasons": reasons,
        "word_count": word_count,
    }

---

Paso 3: Implementar el middleware de routing de modelos

Este es el middleware central del proyecto. Usa el clasificador para decidir qué modelo invocar: gpt-4.1-mini (económico, rápido) para preguntas simples, gpt-4.1 (potente, costoso) para preguntas complejas.

from langchain.agents import AgentMiddleware
from langchain.chat_models import init_chat_model


class ModelRoutingMiddleware(AgentMiddleware):
    """Selecciona el modelo según la complejidad de la pregunta."""

    def __init__(
        self,
        simple_model: str = "openai:gpt-4.1-mini",
        complex_model: str = "openai:gpt-4.1",
    ):
        self._simple = init_chat_model(simple_model, temperature=0)
        self._complex = init_chat_model(complex_model, temperature=0.2)
        self.simple_model_name = simple_model
        self.complex_model_name = complex_model
        self.last_classification = None

    def wrap_model_call(self, messages, config, call_next):
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_user_msg = user_messages[-1].content
            classification = classify_complexity(last_user_msg)
        else:
            classification = {"complexity": "simple", "score": 0, "reasons": ["sin mensaje de usuario"], "word_count": 0}

        self.last_classification = classification

        if classification["complexity"] == "complex":
            selected = self._complex
            model_name = self.complex_model_name
        else:
            selected = self._simple
            model_name = self.simple_model_name

        print(f"\n  🧠 Modelo seleccionado: {model_name}")
        print(f"     Complejidad: {classification['complexity']} (score: {classification['score']})")
        if classification["reasons"]:
            print(f"     Razones: {', '.join(classification['reasons'])}")

        tools_on_messages = []
        for m in messages:
            if hasattr(m, "tool_calls") and m.tool_calls:
                tools_on_messages = m.tool_calls
                break

        bound_tools = config.get("__bound_tools__", [])
        if bound_tools:
            selected = selected.bind_tools(bound_tools)

        return selected.invoke(messages)

---

Paso 4: Agregar el middleware de logging

Este middleware registra cada operación del agente con detalles sobre el modelo usado, los tools ejecutados, y métricas de rendimiento.

import time
from datetime import datetime


class LoggingMiddleware(AgentMiddleware):
    """Registra todas las operaciones con detalle."""

    def __init__(self):
        self.log_entries: list[dict] = []
        self._call_count = 0

    def before_model(self, messages, config):
        self._call_count += 1
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_call",
            "call_number": self._call_count,
            "message_count": len(messages),
        }
        self.log_entries.append(entry)
        print(f"  📋 [{entry['timestamp']}] Llamada #{self._call_count} al modelo ({len(messages)} msgs)")

    def after_model(self, response, config):
        has_tool_calls = bool(getattr(response, "tool_calls", []))
        content_len = len(response.content) if response.content else 0
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_response",
            "has_tool_calls": has_tool_calls,
            "content_length": content_len,
        }
        self.log_entries.append(entry)
        if has_tool_calls:
            tool_names = [tc["name"] for tc in response.tool_calls]
            print(f"  📋 Modelo decidió llamar tools: {', '.join(tool_names)}")
        else:
            preview = response.content[:80] if response.content else "(vacío)"
            print(f"  📋 Modelo respondió: {preview}...")

    def wrap_tool_call(self, tool_call, config, call_next):
        start = time.time()
        print(f"  🔧 Ejecutando: {tool_call['name']}({str(tool_call['args'])[:60]})")
        result = call_next(tool_call, config)
        elapsed = time.time() - start
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "tool_execution",
            "tool": tool_call["name"],
            "elapsed_ms": round(elapsed * 1000),
        }
        self.log_entries.append(entry)
        preview = str(result)[:80]
        print(f"  📥 Resultado ({elapsed * 1000:.0f}ms): {preview}")
        return result

    def get_summary(self) -> dict:
        model_calls = sum(1 for e in self.log_entries if e["type"] == "model_call")
        tool_execs = [e for e in self.log_entries if e["type"] == "tool_execution"]
        tools_used = {}
        for e in tool_execs:
            tools_used[e["tool"]] = tools_used.get(e["tool"], 0) + 1
        return {
            "model_calls": model_calls,
            "tool_executions": len(tool_execs),
            "tools_used": tools_used,
            "total_events": len(self.log_entries),
        }

    def reset(self):
        self.log_entries.clear()
        self._call_count = 0

---

Paso 5: Implementar dynamic tools por complejidad

El middleware de dynamic tools filtra las herramientas disponibles según la clasificación de complejidad. Las preguntas simples solo acceden a search y calculator. Las preguntas complejas acceden a todo: search, calculator, deep_research, y data_analysis.

class DynamicToolsMiddleware(AgentMiddleware):
    """Habilita tools avanzados solo para preguntas complejas."""

    def __init__(self):
        self.basic_tools = [search, calculator]
        self.advanced_tools = [search, calculator, deep_research, data_analysis]

    def get_tools(self, state):
        messages = state.get("messages", [])
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_msg = user_messages[-1].content
            classification = classify_complexity(last_msg)
        else:
            classification = {"complexity": "simple"}

        if classification["complexity"] == "complex":
            tool_names = [t.name for t in self.advanced_tools]
            print(f"  🔧 Tools habilitados (complex): {', '.join(tool_names)}")
            return self.advanced_tools

        tool_names = [t.name for t in self.basic_tools]
        print(f"  🔧 Tools habilitados (simple): {', '.join(tool_names)}")
        return self.basic_tools

---

Paso 6: Componer todo en un agente

Ahora combinamos los tres middleware en el agente final. El orden importa: logging va primero (capa externa, captura todo), luego routing (decide el modelo), y por último dynamic tools (filtra herramientas).

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent

model = init_chat_model("openai:gpt-4.1-mini", temperature=0)

logging_mw = LoggingMiddleware()
routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1",
)
tools_mw = DynamicToolsMiddleware()

SYSTEM_PROMPT = """Eres un asistente inteligente con acceso a herramientas de búsqueda y análisis.

Instrucciones:
- Para preguntas factuales simples, responde de forma concisa
- Para preguntas analíticas complejas, usa las herramientas de investigación profunda y análisis de datos
- Siempre cita tus fuentes cuando uses herramientas de búsqueda
- Si no tienes información suficiente, dilo honestamente"""

agent = create_agent(
    model,
    tools=[search, calculator, deep_research, data_analysis],
    prompt=SYSTEM_PROMPT,
    middleware=[logging_mw, routing_mw, tools_mw],
)

---

Paso 7: Chat loop interactivo con routing visible

El loop interactivo muestra en tiempo real qué modelo se seleccionó, qué tools se habilitaron, y el proceso completo de razonamiento del agente.

def display_routing_decision(routing_mw, question):
    """Muestra la decisión de routing antes de ejecutar."""
    classification = classify_complexity(question)
    print(f"\n{'─' * 60}")
    print(f"  📨 Pregunta: {question}")
    print(f"{'─' * 60}")
    print(f"  📊 Clasificación:")
    print(f"     Complejidad: {classification['complexity'].upper()}")
    print(f"     Score: {classification['score']}")
    if classification["reasons"]:
        for reason in classification["reasons"]:
            print(f"     → {reason}")
    print(f"{'─' * 60}")


def chat_loop():
    """Loop interactivo con routing dinámico visible."""
    print("=" * 60)
    print("  🤖 Agente con Routing Dinámico de Modelos")
    print("  Pregunta lo que quieras — el agente elige el modelo.")
    print("  Comandos: 'stats' (métricas), 'salir' (terminar)")
    print("=" * 60)

    while True:
        try:
            question = input("\n💬 Tu pregunta: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not question:
            continue

        if question.lower() in ("salir", "exit", "quit"):
            print("\n¡Hasta luego!")
            break

        if question.lower() == "stats":
            summary = logging_mw.get_summary()
            print(f"\n📊 Métricas acumuladas:")
            print(f"   Llamadas al modelo: {summary['model_calls']}")
            print(f"   Ejecuciones de tools: {summary['tool_executions']}")
            print(f"   Tools usados: {summary['tools_used']}")
            print(f"   Total eventos: {summary['total_events']}")
            continue

        display_routing_decision(routing_mw, question)

        try:
            result = agent.invoke(
                {"messages": [("user", question)]}
            )

            print(f"\n{'─' * 60}")
            print(f"  💡 Respuesta:")
            print(f"{'─' * 60}")
            print(f"\n{result['messages'][-1].content}")
            print(f"\n{'─' * 60}")

        except Exception as e:
            print(f"\n❌ Error: {e}")
            print("   Intenta con otra pregunta.")


if __name__ == "__main__":
    chat_loop()

---

Código completo

Este es el archivo dynamic_routing_agent.py completo. Cópialo y ejecútalo directamente.

"""
Agente con Routing Dinámico de Modelos
Módulo 4 — LangChain & LangGraph: From Chains to Agents

Selecciona automáticamente el modelo según la complejidad de la pregunta.
"""

from dotenv import load_dotenv
load_dotenv()

import re
import time
from datetime import datetime

from langchain.chat_models import init_chat_model
from langchain.agents import AgentMiddleware, create_agent
from langchain_core.tools import tool


# =============================================================================
# KNOWLEDGE BASE
# =============================================================================

KNOWLEDGE_BASE = {
    "python": {
        "title": "Python Programming Language",
        "content": (
            "Python es un lenguaje de alto nivel, interpretado, con tipado dinámico. "
            "Creado por Guido van Rossum en 1991. Usado en AI/ML, data science, web, "
            "automatización. Ecosistema: PyTorch, TensorFlow, Django, FastAPI. "
            "Community: >8M developers. Versión actual: 3.12."
        ),
        "source": "https://python.org",
    },
    "rust": {
        "title": "Rust Programming Language",
        "content": (
            "Rust es un lenguaje de sistemas con memory safety sin garbage collector. "
            "Creado por Mozilla en 2010. Ownership system único. Performance comparable a C/C++. "
            "Usado en: sistemas operativos, WebAssembly, CLIs, networking. "
            "Adoptado por Linux kernel, Android, Windows."
        ),
        "source": "https://rust-lang.org",
    },
    "microservicios": {
        "title": "Arquitectura de Microservicios",
        "content": (
            "Microservicios es un estilo arquitectónico que estructura una aplicación como "
            "colección de servicios pequeños, autónomos y desplegables independientemente. "
            "Ventajas: escalabilidad independiente, deploy independiente, tecnología heterogénea. "
            "Desventajas: complejidad de red, consistencia eventual, overhead operacional. "
            "Alternativa: monolito modular para equipos < 20 personas."
        ),
        "source": "https://microservices.io",
    },
    "langchain": {
        "title": "LangChain Framework",
        "content": (
            "LangChain es el framework open-source más adoptado para aplicaciones con LLMs. "
            "v1.2+ incluye: init_chat_model, create_agent, middleware system. "
            "Ecosistema: LangGraph (orquestación), LangSmith (observabilidad), Deep Agents. "
            "Usado por: startups AI, enterprise, research labs."
        ),
        "source": "https://langchain.com",
    },
    "typescript": {
        "title": "TypeScript Language",
        "content": (
            "TypeScript es un superset tipado de JavaScript desarrollado por Microsoft. "
            "Agrega tipos estáticos opcionales, interfaces, y enums. Compila a JavaScript. "
            "Adoptado por: Angular, React (con TS), Vue 3, Next.js, Deno. "
            "Market share: >78% de proyectos JavaScript nuevos usan TypeScript."
        ),
        "source": "https://typescriptlang.org",
    },
}


# =============================================================================
# TOOLS — BASIC
# =============================================================================

@tool
def search(query: str) -> str:
    """Busca información general sobre un tema. Útil para preguntas factuales."""
    query_lower = query.lower()
    results = []
    for key, data in KNOWLEDGE_BASE.items():
        if key in query_lower or any(
            word in data["content"].lower() for word in query_lower.split() if len(word) > 3
        ):
            results.append(
                f"📄 {data['title']}: {data['content']} (Fuente: {data['source']})"
            )

    if results:
        return f"Encontré {len(results)} resultado(s):\n" + "\n\n".join(results)
    return (
        f"Sin resultados específicos para '{query}'. "
        f"Temas disponibles: {', '.join(KNOWLEDGE_BASE.keys())}"
    )


@tool
def calculator(expression: str) -> str:
    """Evalúa expresiones matemáticas. Útil para cálculos numéricos."""
    try:
        result = eval(expression)
        return f"Resultado: {expression} = {result}"
    except Exception as e:
        return f"Error evaluando '{expression}': {e}"


# =============================================================================
# TOOLS — ADVANCED (solo para preguntas complejas)
# =============================================================================

@tool
def deep_research(topic: str, aspects: list[str]) -> str:
    """Investigación profunda sobre un tema con múltiples aspectos.
    Solo se usa para preguntas complejas que requieren análisis detallado."""
    results = []
    for aspect in aspects:
        topic_lower = topic.lower()
        if topic_lower in KNOWLEDGE_BASE:
            data = KNOWLEDGE_BASE[topic_lower]
            results.append(f"🔬 {aspect}: Basado en {data['title']} — {data['content']}")
        else:
            results.append(
                f"🔬 {aspect}: Análisis general sobre '{topic}' para el aspecto '{aspect}' "
                f"(sin datos específicos en la base de conocimiento)"
            )

    return (
        f"Investigación profunda sobre '{topic}':\n"
        f"Aspectos analizados: {len(aspects)}\n\n"
        + "\n\n".join(results)
    )


@tool
def data_analysis(data_description: str, analysis_type: str) -> str:
    """Analiza datos con un tipo de análisis específico.
    Solo se usa para preguntas complejas que requieren procesamiento de datos."""
    analyses = {
        "comparison": (
            f"📊 Comparación: {data_description} — Se identificaron 3 dimensiones: "
            f"rendimiento, costo, y facilidad de adopción."
        ),
        "tradeoff": (
            f"📊 Trade-offs: {data_description} — Ventajas vs desventajas en 4 categorías: "
            f"técnica, organizacional, financiera, y temporal."
        ),
        "recommendation": (
            f"📊 Recomendación: {data_description} — La recomendación depende del contexto: "
            f"tamaño del equipo, presupuesto, y timeline del proyecto."
        ),
        "trend": (
            f"📊 Tendencia: {data_description} — Análisis de tendencia muestra crecimiento "
            f"sostenido en los últimos 3 años con aceleración reciente."
        ),
    }
    return analyses.get(
        analysis_type.lower(),
        f"📊 Análisis ({analysis_type}): {data_description} — Completado con resultados generales.",
    )


# =============================================================================
# COMPLEXITY CLASSIFIER
# =============================================================================

COMPLEXITY_KEYWORDS = [
    "analiza", "compara", "evalúa", "explica en detalle",
    "implicaciones", "trade-off", "tradeoff", "ventajas y desventajas",
    "arquitectura", "diseño", "estrategia", "profundidad",
    "vs", "versus", "diferencias entre", "pros y contras",
    "recomendación", "recomienda", "cuándo usar", "mejor opción",
    "impacto", "consecuencias", "a largo plazo",
]


def classify_complexity(text: str) -> dict:
    """Clasifica la complejidad de una pregunta."""
    text_lower = text.lower()
    word_count = len(text.split())

    keyword_matches = [kw for kw in COMPLEXITY_KEYWORDS if kw in text_lower]
    has_multiple_questions = text.count("?") > 1
    has_connectors = any(
        c in text_lower for c in [" y ", " pero ", " sin embargo", " además"]
    )

    score = 0
    reasons = []

    if word_count > 20:
        score += 1
        reasons.append(f"pregunta larga ({word_count} palabras)")
    if keyword_matches:
        score += len(keyword_matches)
        reasons.append(f"keywords analíticas: {', '.join(keyword_matches[:3])}")
    if has_multiple_questions:
        score += 1
        reasons.append("múltiples preguntas")
    if has_connectors:
        score += 1
        reasons.append("conectores complejos")

    complexity = "complex" if score >= 2 else "simple"

    return {
        "complexity": complexity,
        "score": score,
        "reasons": reasons,
        "word_count": word_count,
    }


# =============================================================================
# MIDDLEWARE: Model Routing
# =============================================================================

class ModelRoutingMiddleware(AgentMiddleware):
    """Selecciona el modelo según la complejidad de la pregunta."""

    def __init__(
        self,
        simple_model: str = "openai:gpt-4.1-mini",
        complex_model: str = "openai:gpt-4.1",
    ):
        self._simple = init_chat_model(simple_model, temperature=0)
        self._complex = init_chat_model(complex_model, temperature=0.2)
        self.simple_model_name = simple_model
        self.complex_model_name = complex_model
        self.last_classification = None
        self.routing_history: list[dict] = []

    def wrap_model_call(self, messages, config, call_next):
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_user_msg = user_messages[-1].content
            classification = classify_complexity(last_user_msg)
        else:
            classification = {
                "complexity": "simple",
                "score": 0,
                "reasons": ["sin mensaje de usuario"],
                "word_count": 0,
            }

        self.last_classification = classification

        if classification["complexity"] == "complex":
            selected = self._complex
            model_name = self.complex_model_name
        else:
            selected = self._simple
            model_name = self.simple_model_name

        self.routing_history.append({
            "model": model_name,
            "complexity": classification["complexity"],
            "score": classification["score"],
            "reasons": classification["reasons"],
        })

        print(f"\n  🧠 Modelo seleccionado: {model_name}")
        print(f"     Complejidad: {classification['complexity']} (score: {classification['score']})")
        if classification["reasons"]:
            print(f"     Razones: {', '.join(classification['reasons'])}")

        bound_tools = config.get("__bound_tools__", [])
        if bound_tools:
            selected = selected.bind_tools(bound_tools)

        return selected.invoke(messages)


# =============================================================================
# MIDDLEWARE: Logging
# =============================================================================

class LoggingMiddleware(AgentMiddleware):
    """Registra todas las operaciones con detalle."""

    def __init__(self):
        self.log_entries: list[dict] = []
        self._call_count = 0

    def before_model(self, messages, config):
        self._call_count += 1
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_call",
            "call_number": self._call_count,
            "message_count": len(messages),
        }
        self.log_entries.append(entry)
        print(
            f"  📋 [{entry['timestamp']}] Llamada #{self._call_count} "
            f"al modelo ({len(messages)} msgs)"
        )

    def after_model(self, response, config):
        has_tool_calls = bool(getattr(response, "tool_calls", []))
        content_len = len(response.content) if response.content else 0
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "model_response",
            "has_tool_calls": has_tool_calls,
            "content_length": content_len,
        }
        self.log_entries.append(entry)
        if has_tool_calls:
            tool_names = [tc["name"] for tc in response.tool_calls]
            print(f"  📋 Modelo decidió llamar tools: {', '.join(tool_names)}")
        else:
            preview = response.content[:80] if response.content else "(vacío)"
            print(f"  📋 Modelo respondió: {preview}...")

    def wrap_tool_call(self, tool_call, config, call_next):
        start = time.time()
        print(f"  🔧 Ejecutando: {tool_call['name']}({str(tool_call['args'])[:60]})")

        result = call_next(tool_call, config)

        elapsed = time.time() - start
        entry = {
            "timestamp": datetime.now().strftime("%H:%M:%S"),
            "type": "tool_execution",
            "tool": tool_call["name"],
            "elapsed_ms": round(elapsed * 1000),
        }
        self.log_entries.append(entry)
        preview = str(result)[:80]
        print(f"  📥 Resultado ({elapsed * 1000:.0f}ms): {preview}")
        return result

    def get_summary(self) -> dict:
        model_calls = sum(1 for e in self.log_entries if e["type"] == "model_call")
        tool_execs = [e for e in self.log_entries if e["type"] == "tool_execution"]
        tools_used: dict[str, int] = {}
        for e in tool_execs:
            tools_used[e["tool"]] = tools_used.get(e["tool"], 0) + 1
        return {
            "model_calls": model_calls,
            "tool_executions": len(tool_execs),
            "tools_used": tools_used,
            "total_events": len(self.log_entries),
        }

    def reset(self):
        self.log_entries.clear()
        self._call_count = 0


# =============================================================================
# MIDDLEWARE: Dynamic Tools
# =============================================================================

class DynamicToolsMiddleware(AgentMiddleware):
    """Habilita tools avanzados solo para preguntas complejas."""

    def __init__(self):
        self.basic_tools = [search, calculator]
        self.advanced_tools = [search, calculator, deep_research, data_analysis]

    def get_tools(self, state):
        messages = state.get("messages", [])
        user_messages = [
            m for m in messages
            if hasattr(m, "type") and m.type == "human"
        ]

        if user_messages:
            last_msg = user_messages[-1].content
            classification = classify_complexity(last_msg)
        else:
            classification = {"complexity": "simple"}

        if classification["complexity"] == "complex":
            tool_names = [t.name for t in self.advanced_tools]
            print(f"  🔧 Tools habilitados (complex): {', '.join(tool_names)}")
            return self.advanced_tools

        tool_names = [t.name for t in self.basic_tools]
        print(f"  🔧 Tools habilitados (simple): {', '.join(tool_names)}")
        return self.basic_tools


# =============================================================================
# AGENT SETUP
# =============================================================================

model = init_chat_model("openai:gpt-4.1-mini", temperature=0)

logging_mw = LoggingMiddleware()
routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1",
)
tools_mw = DynamicToolsMiddleware()

SYSTEM_PROMPT = """Eres un asistente inteligente con acceso a herramientas de búsqueda y análisis.

Instrucciones:
- Para preguntas factuales simples, responde de forma concisa
- Para preguntas analíticas complejas, usa las herramientas de investigación profunda y análisis de datos
- Siempre cita tus fuentes cuando uses herramientas de búsqueda
- Si no tienes información suficiente, dilo honestamente"""

agent = create_agent(
    model,
    tools=[search, calculator, deep_research, data_analysis],
    prompt=SYSTEM_PROMPT,
    middleware=[logging_mw, routing_mw, tools_mw],
)


# =============================================================================
# INTERACTIVE CHAT LOOP
# =============================================================================

def display_routing_decision(question):
    """Muestra la clasificación de complejidad."""
    classification = classify_complexity(question)
    print(f"\n{'─' * 60}")
    print(f"  📨 Pregunta: {question}")
    print(f"{'─' * 60}")
    print(f"  📊 Clasificación previa:")
    print(f"     Complejidad: {classification['complexity'].upper()}")
    print(f"     Score: {classification['score']}")
    if classification["reasons"]:
        for reason in classification["reasons"]:
            print(f"     → {reason}")
    print(f"{'─' * 60}")


def chat_loop():
    """Loop interactivo con routing dinámico visible."""
    print("=" * 60)
    print("  🤖 Agente con Routing Dinámico de Modelos")
    print("  Pregunta lo que quieras — el agente elige el modelo.")
    print("  Comandos: 'stats' (métricas), 'salir' (terminar)")
    print("=" * 60)

    while True:
        try:
            question = input("\n💬 Tu pregunta: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not question:
            continue

        if question.lower() in ("salir", "exit", "quit"):
            summary = logging_mw.get_summary()
            print(f"\n📊 Resumen de sesión:")
            print(f"   Llamadas al modelo: {summary['model_calls']}")
            print(f"   Tools ejecutados: {summary['tool_executions']}")
            print(f"   Tools usados: {summary['tools_used']}")
            print(f"\n🧠 Historial de routing:")
            for i, r in enumerate(routing_mw.routing_history, 1):
                print(f"   {i}. {r['model']} ({r['complexity']}, score={r['score']})")
            print("\n¡Hasta luego!")
            break

        if question.lower() == "stats":
            summary = logging_mw.get_summary()
            print(f"\n📊 Métricas acumuladas:")
            print(f"   Llamadas al modelo: {summary['model_calls']}")
            print(f"   Ejecuciones de tools: {summary['tool_executions']}")
            print(f"   Tools usados: {summary['tools_used']}")
            print(f"   Total eventos: {summary['total_events']}")
            print(f"\n🧠 Routing history:")
            for i, r in enumerate(routing_mw.routing_history, 1):
                print(f"   {i}. {r['model']} ({r['complexity']}, score={r['score']})")
            continue

        display_routing_decision(question)

        try:
            result = agent.invoke(
                {"messages": [("user", question)]}
            )

            print(f"\n{'─' * 60}")
            print(f"  💡 Respuesta:")
            print(f"{'─' * 60}")
            print(f"\n{result['messages'][-1].content}")
            print(f"\n{'─' * 60}")

        except Exception as e:
            print(f"\n❌ Error: {e}")
            print("   Intenta con otra pregunta.")


if __name__ == "__main__":
    chat_loop()

Ejecútalo:

python dynamic_routing_agent.py

---

Criterios de éxito

Tu proyecto está completo cuando cumples los cinco criterios:

• ✅ Preguntas simples usan modelo económico — "¿Qué es Python?" usa gpt-4.1-mini, visible en los logs
• ✅ Preguntas complejas usan modelo potente — "Compara microservicios vs monolito y recomienda para una startup" usa gpt-4.1, visible en los logs
• ✅ Logs muestran modelo seleccionado y razón — cada pregunta muestra la clasificación, el score, las razones, y el modelo elegido
• ✅ Tools avanzados solo disponibles para requests complejos — preguntas simples solo acceden a search y calculator; preguntas complejas acceden también a deep_research y data_analysis
• ✅ El sistema funciona como chat interactivo — puedes hacer múltiples preguntas y ver las métricas acumuladas con stats

---

Cómo probar

Test 1: Pregunta simple (debe usar gpt-4.1-mini)

💬 Tu pregunta: ¿Qué es Python?

────────────────────────────────────────────────────────────
  📨 Pregunta: ¿Qué es Python?
────────────────────────────────────────────────────────────
  📊 Clasificación previa:
     Complejidad: SIMPLE
     Score: 0
────────────────────────────────────────────────────────────
  🔧 Tools habilitados (simple): search, calculator
  📋 [15:30:01] Llamada #1 al modelo (2 msgs)

  🧠 Modelo seleccionado: openai:gpt-4.1-mini
     Complejidad: simple (score: 0)
  📋 Modelo decidió llamar tools: search
  🔧 Ejecutando: search({'query': 'Python'})
  📥 Resultado (1ms): Encontré 1 resultado(s): 📄 Python Programming Language...
  📋 [15:30:02] Llamada #2 al modelo (4 msgs)

  🧠 Modelo seleccionado: openai:gpt-4.1-mini
     Complejidad: simple (score: 0)
  📋 Modelo respondió: Python es un lenguaje de programación de alto nivel...

────────────────────────────────────────────────────────────
  💡 Respuesta:
────────────────────────────────────────────────────────────

Python es un lenguaje de programación de alto nivel, interpretado
y con tipado dinámico...

Test 2: Pregunta compleja (debe usar gpt-4.1)

💬 Tu pregunta: Compara las ventajas y desventajas de microservicios vs monolito y recomienda cuándo usar cada uno

────────────────────────────────────────────────────────────
  📨 Pregunta: Compara las ventajas y desventajas de...
────────────────────────────────────────────────────────────
  📊 Clasificación previa:
     Complejidad: COMPLEX
     Score: 4
     → pregunta larga (16 palabras)
     → keywords analíticas: compara, ventajas y desventajas, recomienda
────────────────────────────────────────────────────────────
  🔧 Tools habilitados (complex): search, calculator, deep_research, data_analysis
  📋 [15:31:00] Llamada #3 al modelo (2 msgs)

  🧠 Modelo seleccionado: openai:gpt-4.1
     Complejidad: complex (score: 4)
     Razones: pregunta larga, keywords analíticas: compara, ventajas y desventajas, recomienda
  📋 Modelo decidió llamar tools: deep_research, data_analysis
  🔧 Ejecutando: deep_research({'topic': 'microservicios', 'aspects': [...]})
  📥 Resultado (1ms): Investigación profunda sobre 'microservicios'...
  🔧 Ejecutando: data_analysis({'data_description': '...', 'analysis_type': 'tradeoff'})
  📥 Resultado (0ms): 📊 Trade-offs: ...
  📋 [15:31:03] Llamada #4 al modelo (6 msgs)

  🧠 Modelo seleccionado: openai:gpt-4.1
     Complejidad: complex (score: 4)
  📋 Modelo respondió: [Análisis detallado con recomendaciones]...

────────────────────────────────────────────────────────────
  💡 Respuesta:
────────────────────────────────────────────────────────────

## Microservicios vs Monolito

### Ventajas de Microservicios:
- Escalabilidad independiente...
...

Test 3: Ver métricas

💬 Tu pregunta: stats

📊 Métricas acumuladas:
   Llamadas al modelo: 4
   Ejecuciones de tools: 3
   Tools usados: {'search': 1, 'deep_research': 1, 'data_analysis': 1}
   Total eventos: 10

🧠 Routing history:
   1. openai:gpt-4.1-mini (simple, score=0)
   2. openai:gpt-4.1-mini (simple, score=0)
   3. openai:gpt-4.1 (complex, score=4)
   4. openai:gpt-4.1 (complex, score=4)

Test 4: Pregunta borderline

💬 Tu pregunta: ¿Cuáles son las ventajas de Rust?

────────────────────────────────────────────────────────────
  📊 Clasificación previa:
     Complejidad: SIMPLE
     Score: 0
────────────────────────────────────────────────────────────

(usa gpt-4.1-mini porque "ventajas" sola no es suficiente, necesita "ventajas y desventajas")

---

Errores comunes

1. ModuleNotFoundError: No module named 'langchain'

Causa: No instalaste las dependencias.

pip install langchain langgraph langchain-openai python-dotenv

2. El clasificador marca todo como "simple"

Causa: Las keywords de complejidad no coinciden con la forma en que redactas tus preguntas. El clasificador busca coincidencias exactas.

Solución: Agrega más keywords a COMPLEXITY_KEYWORDS que reflejen tu vocabulario. También puedes ajustar el threshold de score (actualmente >= 2 para complex).

3. AuthenticationError con el modelo potente

Causa: Tu API key no tiene acceso a gpt-4.1. Algunas cuentas de OpenAI solo tienen acceso a modelos mini.

Solución: Cambia el modelo complejo a uno que tengas disponible:

routing_mw = ModelRoutingMiddleware(
    simple_model="openai:gpt-4.1-mini",
    complex_model="openai:gpt-4.1-mini",  # mismo modelo para testing
)

4. El middleware de routing no selecciona el modelo correcto en llamadas intermedias

Causa: Después de la primera llamada al modelo (que decide llamar tools), la segunda llamada ya no tiene el mensaje del usuario como último mensaje — tiene los resultados de tools. El clasificador ve estos mensajes técnicos y los clasifica como "simple".

Solución: El middleware busca el último mensaje con type == "human", no el último mensaje. Verifica que la lógica de user_messages filtra correctamente por tipo.

5. Los tools avanzados aparecen en preguntas simples

Causa: DynamicToolsMiddleware.get_tools() y la clasificación del router pueden divergir si el estado de mensajes cambia entre las dos evaluaciones.

Solución: Ambos middleware usan classify_complexity() con la misma lógica, así que deberían coincidir. Si ves inconsistencias, verifica que el último mensaje del usuario sea el mismo en ambos casos.

6. RateLimitError por demasiadas llamadas

Causa: Cada pregunta genera 2-4 llamadas al modelo (clasificación + tool calling + respuesta). Las preguntas complejas generan más porque usan tools avanzados que retornan más contexto.

Solución: Agrega un RateLimitMiddleware (lo viste en la cápsula 07) a la lista de middleware, o espera entre preguntas.

7. El agente no usa los tools avanzados aunque la pregunta es compleja

Causa: El system prompt no es lo suficientemente directivo para usar deep_research y data_analysis. El modelo puede decidir responder directamente.

Solución: Haz el system prompt más explícito: "Para preguntas complejas que requieren análisis, SIEMPRE usa deep_research y/o data_analysis antes de responder."

8. TypeError: 'NoneType' object is not subscriptable en result["messages"]

Causa: El agente retornó None, usualmente porque una excepción fue silenciada dentro del middleware.

Solución: Verifica que todos tus wrap_model_call y wrap_tool_call retornan un valor. Olvidar el return es el error más común.

---

Ideas para extender

Si terminaste el proyecto y quieres ir más allá:

• Clasificador con LLM — Reemplaza la heurística de keywords con una llamada rápida a gpt-4.1-mini que clasifique la complejidad. Es más costoso pero mucho más preciso para casos ambiguos
• Tres niveles de modelo — Agrega un nivel intermedio: gpt-4.1-mini para simple, gpt-4.1-mini con temperature alta para medio, y gpt-4.1 para complejo. Ajusta el classifier con tres thresholds
• Presupuesto de costo — Agrega un BudgetMiddleware que trackee el costo estimado (input tokens × precio por token) y cambie al modelo económico cuando se exceda un presupuesto diario
• Más proveedores — Agrega Anthropic como opción: gpt-4.1-mini para simple, claude-sonnet para complejo. Requiere pip install langchain-anthropic y ANTHROPIC_API_KEY
• Feedback loop — Permite que el usuario diga "esa respuesta fue insuficiente" y el agente re-ejecute con el modelo potente. Implementa como un comando especial en el chat loop
• Persistencia — Agrega MemorySaver para que el agente recuerde conversaciones previas y use el historial para mejorar la clasificación de complejidad

---

Conexión con el siguiente módulo

En este proyecto usaste todo el sistema de middleware de LangChain para crear un agente inteligente que optimiza costo y calidad. Pero todo funciona dentro de create_agent — un loop lineal de model → tools → model → tools → respuesta.

¿Qué pasa si necesitas que el agente tome caminos completamente diferentes según la situación? ¿Que un nodo procese datos mientras otro busca información en paralelo? ¿Que el flujo haga un loop de retry solo en caso de error? ¿Que se pause para pedir aprobación humana?

En el Módulo 5: Introducción a LangGraph, aprenderás a construir workflows como grafos: nodos que son funciones, edges que definen el flujo, y conditional edges que crean branching dinámico. LangGraph te da control total sobre la orquestación — incluyendo ciclos, branching, subgraphs, y persistencia de estado — que van mucho más allá de lo que middleware puede lograr. El create_agent que usaste en este módulo es, de hecho, un grafo de LangGraph bajo el capó. En el Módulo 5, aprenderás a construir esos grafos tú mismo.

---

Recursos del proyecto

1. LangChain Agents Middleware — Guía oficial del sistema de middleware
2. create_agent API Reference — Referencia con parámetros de middleware
3. LangChain init_chat_model — Inicialización multi-proveedor para routing
4. OpenAI Model Pricing — Precios de modelos para entender el ahorro del routing
5. LangGraph Agents Conceptual Guide — Cómo create_agent funciona internamente como un grafo
6. Middleware Pattern in Software — El patrón de diseño detrás del middleware system

---

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