Módulo 4: Middleware y Customización

AgentMiddleware Class: Middleware Compuesto

Descripción de la cápsula

En las cápsulas anteriores aprendiste a personalizar agentes con hooks individuales: @before_model para interceptar antes de la llamada al modelo, @after_model para actuar después de la respuesta, @wrap_model_call para envolver la llamada completa, y @wrap_tool_call para controlar la ejecución de tools. También viste cómo usar dynamic models, dynamic tools, y dynamic prompts para cambiar el comportamiento del agente en runtime.

Pero hay un problema que aparece rápido en proyectos reales: los hooks están sueltos. Tu archivo tiene un @before_model para logging por aquí, un @wrap_tool_call para auth por allá, un @wrap_model_call para routing de modelos más abajo. Cuando tienes un segundo agente que necesita el mismo logging, copias y pegas. Cuando un tercero necesita logging + auth pero no routing, copias selectivamente. En pocas semanas, tienes hooks duplicados, inconsistentes, y difíciles de mantener.

La clase AgentMiddleware resuelve esto. Te permite empaquetar múltiples hooks, estado custom, y tools en una sola unidad reutilizable — como un plugin que puedes instalar en cualquier agente. Defines una clase que hereda de AgentMiddleware, implementas los hooks que necesitas como métodos, y la pasas al agente. Cuando tienes múltiples middleware, los compones en una lista: middleware=[LoggingMiddleware(), AuthMiddleware(), CachingMiddleware()]. LangChain los ejecuta en orden, creando capas de comportamiento que se apilan sin conflicto.

El problema: hooks individuales no escalan

Imagina que tienes tres agentes en tu sistema y cada uno necesita combinaciones diferentes de comportamiento:

Agente	Logging	Auth	Rate limiting
Agente de soporte	✅	✅	✅
Agente interno	✅	❌	❌
Agente de análisis	✅	✅	❌

Con hooks individuales, tu código se ve así:

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

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

# Agente de soporte: necesita logging + auth + rate limiting
support_agent = create_agent(
    model, [search],
    prompt="Eres un agente de soporte.",
    before_model=log_before,       # ← hook suelto
    after_model=log_after,         # ← hook suelto
    wrap_tool_call=auth_tool,      # ← hook suelto
    wrap_model_call=rate_limit,    # ← hook suelto
)

# Agente interno: solo logging
internal_agent = create_agent(
    model, [search],
    prompt="Eres un agente interno.",
    before_model=log_before,       # ← copiado
    after_model=log_after,         # ← copiado
)

# Agente de análisis: logging + auth
analysis_agent = create_agent(
    model, [search],
    prompt="Eres un agente de análisis.",
    before_model=log_before,       # ← copiado de nuevo
    after_model=log_after,         # ← copiado de nuevo
    wrap_tool_call=auth_tool,      # ← copiado
)

Tres agentes, y ya tienes hooks copiados 6 veces. Si quieres cambiar cómo funciona el logging, tienes que actualizar cada agente. Si agregas un cuarto agente con logging + rate limiting, copias de nuevo. Este patrón no escala.

AgentMiddleware class: la solución

AgentMiddleware es una clase base que combina hooks, estado, y tools en una sola unidad. En vez de pasar hooks individuales, creas clases que encapsulan comportamiento y las pasas como una lista al agente.

Estructura básica

from dotenv import load_dotenv
load_dotenv()

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

class LoggingMiddleware(AgentMiddleware):
    """Registra todas las operaciones del agente."""

    def before_model(self, messages, config):
        print(f"[LOG] Llamada al modelo con {len(messages)} mensajes")

    def after_model(self, response, config):
        preview = response.content[:80] if response.content else "(sin contenido)"
        print(f"[LOG] Modelo respondió: {preview}")

    def wrap_tool_call(self, tool_call, config, call_next):
        print(f"[LOG] Ejecutando tool: {tool_call['name']}")
        result = call_next(tool_call, config)
        preview = str(result)[:80]
        print(f"[LOG] Resultado: {preview}")
        return result

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"Resultados sobre {query}: Python es un lenguaje versátil..."

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente de investigación.",
    middleware=[LoggingMiddleware()],
)

result = agent.invoke(
    {"messages": [("user", "¿Qué es Python?")]}
)
print(result["messages"][-1].content)
# Output esperado:
# [LOG] Llamada al modelo con 2 mensajes
# [LOG] Ejecutando tool: search
# [LOG] Resultado: Resultados sobre Python: Python es un lenguaje versátil...
# [LOG] Llamada al modelo con 4 mensajes
# [LOG] Modelo respondió: Python es un lenguaje de programación de alto nivel...
# Python es un lenguaje de programación de alto nivel...

Cada método es opcional. Solo implementas los hooks que necesitas:

✅ before_model(messages, config) — se ejecuta antes de cada llamada al modelo
✅ after_model(response, config) — se ejecuta después de cada respuesta del modelo
✅ wrap_model_call(messages, config, call_next) — envuelve la llamada completa al modelo
✅ wrap_tool_call(tool_call, config, call_next) — envuelve la ejecución de cada tool
✅ get_tools(state) — retorna tools dinámicos según el estado
✅ get_prompt(state) — retorna prompt dinámico según el estado

Crear una middleware class paso a paso

Vamos a construir un middleware de autenticación que valida permisos antes de ejecutar tools y filtra qué tools están disponibles según el rol del usuario.

Paso 1: definir el middleware

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información general."""
    return f"Resultados para: {query}"

@tool
def calculator(expression: str) -> str:
    """Evalúa expresiones matemáticas."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

@tool
def admin_panel(action: str) -> str:
    """Ejecuta acciones administrativas. Solo para admins."""
    return f"Acción administrativa ejecutada: {action}"

class AuthMiddleware(AgentMiddleware):
    """Middleware que controla acceso por rol de usuario."""

    def get_tools(self, state):
        user_role = state.get("user_role", "basic")
        if user_role == "admin":
            return [search, calculator, admin_panel]
        return [search, calculator]

    def wrap_tool_call(self, tool_call, config, call_next):
        state = config.get("configurable", {})
        user_role = state.get("user_role", "basic")
        tool_name = tool_call["name"]

        if tool_name == "admin_panel" and user_role != "admin":
            return "Error: No tienes permisos para ejecutar acciones administrativas."

        print(f"[AUTH] Usuario ({user_role}) ejecutando: {tool_name}")
        return call_next(tool_call, config)

Paso 2: usar el middleware en un agente

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_role: str

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

agent = create_agent(
    model,
    tools=[search, calculator, admin_panel],
    prompt="Eres un asistente. Usa las herramientas disponibles para ayudar al usuario.",
    state_schema=AgentState,
    middleware=[AuthMiddleware()],
)

# Usuario básico
result_basic = agent.invoke({
    "messages": [("user", "Ejecuta una acción administrativa")],
    "user_role": "basic",
})
print(result_basic["messages"][-1].content)
# Output esperado:
# [AUTH] Usuario (basic) ejecutando: admin_panel
# No tengo permisos para ejecutar acciones administrativas...

# Usuario admin
result_admin = agent.invoke({
    "messages": [("user", "Ejecuta una acción administrativa de respaldo")],
    "user_role": "admin",
})
print(result_admin["messages"][-1].content)
# Output esperado:
# [AUTH] Usuario (admin) ejecutando: admin_panel
# La acción administrativa de respaldo fue ejecutada exitosamente.

El middleware filtra tools en get_tools (el usuario básico ni siquiera ve admin_panel) y además valida en wrap_tool_call como segunda capa de seguridad.

Componer múltiples middleware

La composición es donde AgentMiddleware brilla. Pasas una lista de middleware y LangChain los ejecuta en orden — el primer middleware de la lista es la capa más externa.

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class LoggingMiddleware(AgentMiddleware):
    """Capa 1: Registra todas las operaciones."""

    def wrap_model_call(self, messages, config, call_next):
        start = time.time()
        print(f"[LOG] → Llamada al modelo ({len(messages)} msgs)")
        response = call_next(messages, config)
        elapsed = time.time() - start
        print(f"[LOG] ← Respuesta en {elapsed:.2f}s")
        return response

    def wrap_tool_call(self, tool_call, config, call_next):
        print(f"[LOG] 🔧 Tool: {tool_call['name']}")
        result = call_next(tool_call, config)
        print(f"[LOG] 📥 Resultado: {str(result)[:60]}")
        return result

class CachingMiddleware(AgentMiddleware):
    """Capa 2: Cache de respuestas de tools."""

    def __init__(self):
        self._cache = {}

    def wrap_tool_call(self, tool_call, config, call_next):
        cache_key = f"{tool_call['name']}:{tool_call['args']}"
        if cache_key in self._cache:
            print(f"[CACHE] Hit para {tool_call['name']}")
            return self._cache[cache_key]

        result = call_next(tool_call, config)
        self._cache[cache_key] = result
        print(f"[CACHE] Almacenado: {tool_call['name']}")
        return result

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente. Busca información cuando te pregunten.",
    middleware=[LoggingMiddleware(), CachingMiddleware()],
)

result = agent.invoke(
    {"messages": [("user", "Busca sobre Python y luego sobre Python otra vez")]}
)
print(result["messages"][-1].content)
# Output esperado:
# [LOG] → Llamada al modelo (2 msgs)
# [LOG] ← Respuesta en 1.23s
# [LOG] 🔧 Tool: search
# [CACHE] Almacenado: search
# [LOG] 📥 Resultado: Resultados para: Python
# [LOG] → Llamada al modelo (4 msgs)
# [LOG] ← Respuesta en 0.89s
# [LOG] 🔧 Tool: search
# [CACHE] Hit para search
# [LOG] 📥 Resultado: Resultados para: Python
# ...

El orden importa: cómo se apilan los middleware

Cuando pasas middleware=[A(), B(), C()], LangChain los ejecuta así:

A.wrap_model_call →
    B.wrap_model_call →
        C.wrap_model_call →
            modelo real
        ← C retorna
    ← B retorna
← A retorna

El primer middleware de la lista es la capa más externa (la primera en ejecutarse y la última en recibir el resultado). Esto importa cuando el orden afecta el comportamiento:

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class TimerMiddleware(AgentMiddleware):
    """Mide tiempo total (incluyendo el tiempo del middleware interno)."""

    def wrap_model_call(self, messages, config, call_next):
        import time
        start = time.time()
        response = call_next(messages, config)
        elapsed = time.time() - start
        print(f"[TIMER] Tiempo total (modelo + middleware internos): {elapsed:.2f}s")
        return response

class RetryMiddleware(AgentMiddleware):
    """Reintenta si el modelo falla."""

    def wrap_model_call(self, messages, config, call_next):
        max_retries = 3
        for attempt in range(max_retries):
            try:
                return call_next(messages, config)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                print(f"[RETRY] Intento {attempt + 1} falló: {e}. Reintentando...")

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

# Timer es capa externa → mide tiempo total incluyendo retries
agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[TimerMiddleware(), RetryMiddleware()],
)

result = agent.invoke(
    {"messages": [("user", "Busca sobre Python")]}
)
# Output esperado:
# [TIMER] Tiempo total (modelo + middleware internos): 1.45s

Si inviertes el orden ([RetryMiddleware(), TimerMiddleware()]), el timer mediría solo el tiempo del modelo sin incluir los retries. El orden que elijas depende de lo que quieras medir o controlar.

Regla práctica de ordenamiento:

Posición	Tipo de middleware	Por qué
Primero (externo)	Logging / Métricas	Captura todo, incluyendo retries y errores
Medio	Auth / Validación	Bloquea antes de que llegue al modelo
Último (interno)	Retry / Cache	Actúa directamente sobre la llamada al modelo

Middleware como módulos reutilizables

La verdadera ventaja es que un AgentMiddleware es una clase Python normal — puedes parametrizarla, heredarla, y compartirla entre proyectos.

Middleware parametrizable

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class ConfigurableLoggingMiddleware(AgentMiddleware):
    """Logging con nivel configurable."""

    def __init__(self, level: str = "info", log_tools: bool = True, log_model: bool = True):
        self.level = level
        self.log_tools = log_tools
        self.log_model = log_model

    def before_model(self, messages, config):
        if self.log_model:
            print(f"[{self.level.upper()}] Modelo: {len(messages)} mensajes")

    def wrap_tool_call(self, tool_call, config, call_next):
        if self.log_tools:
            print(f"[{self.level.upper()}] Tool: {tool_call['name']}")
        return call_next(tool_call, config)

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

# Desarrollo: logging completo
dev_agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[ConfigurableLoggingMiddleware(level="debug", log_tools=True, log_model=True)],
)

# Producción: solo tools
prod_agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[ConfigurableLoggingMiddleware(level="warn", log_tools=True, log_model=False)],
)

result = dev_agent.invoke({"messages": [("user", "Busca Python")]})
# Output esperado:
# [DEBUG] Modelo: 2 mensajes
# [DEBUG] Tool: search
# [DEBUG] Modelo: 4 mensajes

Middleware con estado interno

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class MetricsMiddleware(AgentMiddleware):
    """Recolecta métricas de uso."""

    def __init__(self):
        self.model_calls = 0
        self.tool_calls = 0
        self.total_input_messages = 0

    def before_model(self, messages, config):
        self.model_calls += 1
        self.total_input_messages += len(messages)

    def wrap_tool_call(self, tool_call, config, call_next):
        self.tool_calls += 1
        return call_next(tool_call, config)

    def get_metrics(self) -> dict:
        return {
            "model_calls": self.model_calls,
            "tool_calls": self.tool_calls,
            "total_input_messages": self.total_input_messages,
        }

metrics = MetricsMiddleware()

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente de investigación.",
    middleware=[metrics],
)

result = agent.invoke({"messages": [("user", "Busca sobre Python y TypeScript")]})

print(metrics.get_metrics())
# Output esperado:
# {'model_calls': 2, 'tool_calls': 2, 'total_input_messages': 6}

Como metrics es una instancia que mantiene estado, puedes consultar las métricas después de cada invocación. Esto es especialmente útil para dashboards de monitoreo.

Patrones built-in comunes

Estos son los patrones de middleware que aparecen en casi todo proyecto de producción. Puedes usarlos como base y adaptarlos.

Rate limiting middleware

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class RateLimitMiddleware(AgentMiddleware):
    """Limita la frecuencia de llamadas al modelo."""

    def __init__(self, max_calls_per_minute: int = 20):
        self.max_calls = max_calls_per_minute
        self._timestamps: list[float] = []

    def wrap_model_call(self, messages, config, call_next):
        now = time.time()
        self._timestamps = [t for t in self._timestamps if now - t < 60]

        if len(self._timestamps) >= self.max_calls:
            wait_time = 60 - (now - self._timestamps[0])
            print(f"[RATE] Límite alcanzado. Esperando {wait_time:.1f}s...")
            time.sleep(wait_time)

        self._timestamps.append(time.time())
        return call_next(messages, config)

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[RateLimitMiddleware(max_calls_per_minute=10)],
)

result = agent.invoke({"messages": [("user", "Busca sobre Python")]})
print(result["messages"][-1].content)
# Output esperado (sin rate limit hit):
# Python es un lenguaje de programación...

Error recovery middleware

from dotenv import load_dotenv
load_dotenv()

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

@tool
def unreliable_api(query: str) -> str:
    """API que puede fallar."""
    import random
    if random.random() < 0.3:
        raise ConnectionError("API temporalmente no disponible")
    return f"Datos sobre: {query}"

class ErrorRecoveryMiddleware(AgentMiddleware):
    """Maneja errores de tools con retry y fallback."""

    def __init__(self, max_retries: int = 2, fallback_message: str = "Servicio no disponible"):
        self.max_retries = max_retries
        self.fallback_message = fallback_message

    def wrap_tool_call(self, tool_call, config, call_next):
        for attempt in range(self.max_retries + 1):
            try:
                return call_next(tool_call, config)
            except Exception as e:
                if attempt < self.max_retries:
                    print(f"[RECOVERY] {tool_call['name']} falló (intento {attempt + 1}): {e}")
                    continue
                print(f"[RECOVERY] {tool_call['name']} falló definitivamente. Usando fallback.")
                return self.fallback_message

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

agent = create_agent(
    model, [unreliable_api],
    prompt="Eres un asistente. Busca información cuando te pregunten.",
    middleware=[ErrorRecoveryMiddleware(max_retries=2)],
)

result = agent.invoke({"messages": [("user", "Busca datos sobre ML")]})
print(result["messages"][-1].content)
# Output esperado (con retry exitoso):
# [RECOVERY] unreliable_api falló (intento 1): API temporalmente no disponible
# ML (Machine Learning) es una rama de la inteligencia artificial...

Crear una librería de middleware para tu organización

Cuando tienes middleware estables que se usan en múltiples proyectos, el siguiente paso es empaquetarlos como un módulo Python que cualquier equipo puede importar.

Estructura recomendada

my_middleware/
├── __init__.py
├── logging.py
├── auth.py
├── rate_limiting.py
├── caching.py
└── metrics.py

Ejemplo de módulo compartido

# my_middleware/__init__.py
from .logging import LoggingMiddleware
from .auth import AuthMiddleware
from .rate_limiting import RateLimitMiddleware
from .caching import CachingMiddleware
from .metrics import MetricsMiddleware

__all__ = [
    "LoggingMiddleware",
    "AuthMiddleware",
    "RateLimitMiddleware",
    "CachingMiddleware",
    "MetricsMiddleware",
]

# my_middleware/logging.py
from langchain.agents import AgentMiddleware

class LoggingMiddleware(AgentMiddleware):
    """Logging estandarizado para todos los agentes de la organización."""

    def __init__(self, service_name: str = "default", log_level: str = "info"):
        self.service_name = service_name
        self.log_level = log_level

    def before_model(self, messages, config):
        print(f"[{self.service_name}] [{self.log_level.upper()}] "
              f"Model call: {len(messages)} messages")

    def after_model(self, response, config):
        token_count = getattr(response, "usage_metadata", {})
        print(f"[{self.service_name}] [{self.log_level.upper()}] "
              f"Model response: {len(response.content)} chars")

    def wrap_tool_call(self, tool_call, config, call_next):
        print(f"[{self.service_name}] Tool: {tool_call['name']}")
        result = call_next(tool_call, config)
        print(f"[{self.service_name}] Tool done: {tool_call['name']}")
        return result

# Uso en cualquier proyecto de la organizació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

from my_middleware import LoggingMiddleware, RateLimitMiddleware

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[
        LoggingMiddleware(service_name="support-bot", log_level="debug"),
        RateLimitMiddleware(max_calls_per_minute=30),
    ],
)

result = agent.invoke({"messages": [("user", "Busca sobre Python")]})
# Output esperado:
# [support-bot] [DEBUG] Model call: 2 messages
# [support-bot] [DEBUG] Model response: 156 chars
# [support-bot] Tool: search
# [support-bot] Tool done: search

Troubleshooting

1. Los hooks del middleware no se ejecutan

Causa: El nombre del método no coincide exactamente con el esperado. Los nombres válidos son before_model, after_model, wrap_model_call, wrap_tool_call, get_tools, get_prompt.

Solución: Verifica que los nombres de los métodos sean exactos. Python no te avisará si defines before_Model (con M mayúscula) — simplemente no se ejecutará.

2. `wrap_model_call` y `before_model` se ejecutan los dos

Causa: Esto es comportamiento correcto. before_model se ejecuta primero, luego wrap_model_call. No son mutuamente excluyentes.

Solución: Si solo necesitas uno, implementa solo ese. Usa before_model / after_model para side effects simples (logging, métricas). Usa wrap_model_call cuando necesitas modificar la llamada o su respuesta.

3. El middleware con estado se comparte entre invocaciones

Causa: Si usas la misma instancia de middleware para múltiples agentes o invocaciones, el estado (self._cache, self.model_calls, etc.) se acumula.

Solución: Esto es intencional para middleware como MetricsMiddleware (quieres acumular métricas). Para middleware donde quieres estado fresco, crea una nueva instancia por agente o implementa un método reset().

4. `call_next` no está disponible en `before_model`

Causa: before_model y after_model son hooks de side effect — no envuelven la llamada. Solo wrap_model_call y wrap_tool_call reciben call_next.

Solución: Si necesitas modificar la llamada al modelo (no solo observarla), usa wrap_model_call en vez de before_model.

5. El orden de middleware no produce el resultado esperado

Causa: Recuerda que el primer middleware en la lista es la capa más externa. Si A está antes que B, A.wrap_model_call envuelve a B.wrap_model_call.

Solución: Dibuja el flujo como capas de cebolla: [exterior → interior]. Logging va primero (captura todo), retry va último (directamente sobre el modelo).

Ejercicios

Ejercicio 1: Middleware básico de timestamp (Básico)

Crea un TimestampMiddleware que imprima la fecha y hora de cada llamada al modelo y de cada ejecución de tool.

Ver solución

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class TimestampMiddleware(AgentMiddleware):
    """Agrega timestamps a todas las operaciones."""

    def before_model(self, messages, config):
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{now}] Model call ({len(messages)} msgs)")

    def wrap_tool_call(self, tool_call, config, call_next):
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{now}] Tool: {tool_call['name']}")
        result = call_next(tool_call, config)
        after = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{after}] Tool completado: {tool_call['name']}")
        return result

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[TimestampMiddleware()],
)

result = agent.invoke({"messages": [("user", "Busca sobre Python")]})
print(result["messages"][-1].content)
# Output esperado:
# [2026-02-28 15:30:00] Model call (2 msgs)
# [2026-02-28 15:30:01] Tool: search
# [2026-02-28 15:30:01] Tool completado: search
# [2026-02-28 15:30:01] Model call (4 msgs)
# Python es un lenguaje de programación...

Ejercicio 2: Middleware de conteo de tokens (Básico)

Crea un TokenCounterMiddleware que cuente cuántas llamadas al modelo se hicieron y cuántos tools se ejecutaron. El middleware debe tener un método report() que imprima un resumen.

Ver solución

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

@tool
def calculator(expression: str) -> str:
    """Evalúa una expresión matemática."""
    return str(eval(expression))

class TokenCounterMiddleware(AgentMiddleware):
    """Cuenta llamadas al modelo y ejecuciones de tools."""

    def __init__(self):
        self.model_calls = 0
        self.tool_calls = 0
        self.tools_used: dict[str, int] = {}

    def before_model(self, messages, config):
        self.model_calls += 1

    def wrap_tool_call(self, tool_call, config, call_next):
        self.tool_calls += 1
        name = tool_call["name"]
        self.tools_used[name] = self.tools_used.get(name, 0) + 1
        return call_next(tool_call, config)

    def report(self):
        print(f"\n📊 Reporte de uso:")
        print(f"   Llamadas al modelo: {self.model_calls}")
        print(f"   Ejecuciones de tools: {self.tool_calls}")
        print(f"   Tools usadas: {self.tools_used}")

counter = TokenCounterMiddleware()
model = init_chat_model("openai:gpt-4.1-mini")

agent = create_agent(
    model, [search, calculator],
    prompt="Eres un asistente de investigación.",
    middleware=[counter],
)

result = agent.invoke(
    {"messages": [("user", "Busca sobre Python y calcula 2**10")]}
)
counter.report()
# Output esperado:
# 📊 Reporte de uso:
#    Llamadas al modelo: 2
#    Ejecuciones de tools: 2
#    Tools usadas: {'search': 1, 'calculator': 1}

Ejercicio 3: Componer logging + rate limiting (Intermedio)

Crea dos middleware (SimpleLogger y SimpleRateLimiter) y compónlos en un agente. El rate limiter debe esperar 1 segundo entre llamadas al modelo. El logger debe mostrar que el wait ocurre.

Ver solución

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class SimpleLogger(AgentMiddleware):
    """Logging con timestamps."""

    def wrap_model_call(self, messages, config, call_next):
        start = time.time()
        print(f"[LOG] → Model call iniciado")
        response = call_next(messages, config)
        elapsed = time.time() - start
        print(f"[LOG] ← Model call completado ({elapsed:.2f}s)")
        return response

class SimpleRateLimiter(AgentMiddleware):
    """Espera entre llamadas al modelo."""

    def __init__(self, min_interval: float = 1.0):
        self.min_interval = min_interval
        self._last_call = 0.0

    def wrap_model_call(self, messages, config, call_next):
        now = time.time()
        elapsed = now - self._last_call
        if elapsed < self.min_interval and self._last_call > 0:
            wait = self.min_interval - elapsed
            print(f"[RATE] Esperando {wait:.2f}s...")
            time.sleep(wait)
        self._last_call = time.time()
        return call_next(messages, config)

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

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[SimpleLogger(), SimpleRateLimiter()],
)

result = agent.invoke({"messages": [("user", "Busca sobre Python")]})
print(result["messages"][-1].content)
# Output esperado:
# [LOG] → Model call iniciado
# [LOG] ← Model call completado (1.12s)
# [LOG] → Model call iniciado
# [RATE] Esperando 0.88s...
# [LOG] ← Model call completado (2.01s)

Ejercicio 4: Middleware de auditoría con historial (Intermedio)

Crea un AuditMiddleware que almacene un historial de todas las operaciones (modelo y tools) con timestamp, tipo, y detalles. Debe tener un método get_audit_log() que retorne la lista completa.

Ver solución

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class AuditMiddleware(AgentMiddleware):
    """Registra un log de auditoría de todas las operaciones."""

    def __init__(self):
        self._log: list[dict] = []

    def before_model(self, messages, config):
        self._log.append({
            "timestamp": datetime.now().isoformat(),
            "type": "model_call",
            "details": {"message_count": len(messages)},
        })

    def after_model(self, response, config):
        self._log.append({
            "timestamp": datetime.now().isoformat(),
            "type": "model_response",
            "details": {
                "content_length": len(response.content) if response.content else 0,
                "has_tool_calls": bool(getattr(response, "tool_calls", [])),
            },
        })

    def wrap_tool_call(self, tool_call, config, call_next):
        self._log.append({
            "timestamp": datetime.now().isoformat(),
            "type": "tool_call",
            "details": {
                "tool": tool_call["name"],
                "args": tool_call["args"],
            },
        })
        result = call_next(tool_call, config)
        self._log.append({
            "timestamp": datetime.now().isoformat(),
            "type": "tool_result",
            "details": {
                "tool": tool_call["name"],
                "result_length": len(str(result)),
            },
        })
        return result

    def get_audit_log(self) -> list[dict]:
        return self._log

audit = AuditMiddleware()
model = init_chat_model("openai:gpt-4.1-mini")

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[audit],
)

result = agent.invoke({"messages": [("user", "Busca sobre Rust")]})

for entry in audit.get_audit_log():
    print(f"  [{entry['type']}] {entry['details']}")
# Output esperado:
#   [model_call] {'message_count': 2}
#   [model_response] {'content_length': 0, 'has_tool_calls': True}
#   [tool_call] {'tool': 'search', 'args': {'query': 'Rust'}}
#   [tool_result] {'tool': 'search', 'result_length': 24}
#   [model_call] {'message_count': 4}
#   [model_response] {'content_length': 185, 'has_tool_calls': False}

Ejercicio 5: Middleware que cambia el modelo por costo (Avanzado)

Crea un CostAwareMiddleware que use gpt-4.1-mini cuando el input tiene menos de 100 palabras y gpt-4.1 cuando tiene más. Debe registrar qué modelo se usó en cada llamada.

Ver solución

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search(query: str) -> str:
    """Busca información."""
    return f"Resultados para: {query}"

class CostAwareMiddleware(AgentMiddleware):
    """Selecciona modelo según la longitud del input."""

    def __init__(self):
        self.model_log: list[dict] = []
        self._mini = init_chat_model("openai:gpt-4.1-mini")
        self._full = init_chat_model("openai:gpt-4.1")

    def wrap_model_call(self, messages, config, call_next):
        total_words = sum(
            len(str(m.content).split())
            for m in messages
            if hasattr(m, "content") and m.content
        )

        if total_words < 100:
            model_name = "gpt-4.1-mini"
            selected_model = self._mini
        else:
            model_name = "gpt-4.1"
            selected_model = self._full

        self.model_log.append({
            "model": model_name,
            "word_count": total_words,
            "reason": "short input" if total_words < 100 else "long input",
        })
        print(f"[COST] Usando {model_name} ({total_words} palabras)")

        return selected_model.invoke(messages)

model = init_chat_model("openai:gpt-4.1-mini")
cost_middleware = CostAwareMiddleware()

agent = create_agent(
    model, [search],
    prompt="Eres un asistente.",
    middleware=[cost_middleware],
)

result = agent.invoke({"messages": [("user", "Hola")]})
print(result["messages"][-1].content)
print(f"\nModelos usados: {cost_middleware.model_log}")
# Output esperado:
# [COST] Usando gpt-4.1-mini (3 palabras)
# ¡Hola! ¿En qué puedo ayudarte?
# Modelos usados: [{'model': 'gpt-4.1-mini', 'word_count': 3, 'reason': 'short input'}]

Ejercicio 6: Librería completa con 3 middleware compuestos (Challenge)

Crea tres middleware (LoggerMW, AuthMW, MetricsMW), compónlos en un agente con state_schema que incluya user_role, y demuestra que:

El logger registra todo
El auth filtra tools por rol
Las métricas se acumulan correctamente

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain.agents import AgentMiddleware, create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Busca información general."""
    return f"Resultados: {query}"

@tool
def admin_action(action: str) -> str:
    """Ejecuta acción administrativa (solo admins)."""
    return f"Admin: {action} ejecutado"

class LoggerMW(AgentMiddleware):
    def __init__(self):
        self.logs: list[str] = []

    def before_model(self, messages, config):
        entry = f"MODEL_CALL({len(messages)} msgs)"
        self.logs.append(entry)
        print(f"[LOG] {entry}")

    def wrap_tool_call(self, tool_call, config, call_next):
        entry = f"TOOL({tool_call['name']})"
        self.logs.append(entry)
        print(f"[LOG] {entry}")
        return call_next(tool_call, config)

class AuthMW(AgentMiddleware):
    def get_tools(self, state):
        role = state.get("user_role", "basic")
        if role == "admin":
            return [search, admin_action]
        return [search]

class MetricsMW(AgentMiddleware):
    def __init__(self):
        self.calls = {"model": 0, "tools": 0}

    def before_model(self, messages, config):
        self.calls["model"] += 1

    def wrap_tool_call(self, tool_call, config, call_next):
        self.calls["tools"] += 1
        return call_next(tool_call, config)

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    user_role: str

logger = LoggerMW()
metrics = MetricsMW()

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

agent = create_agent(
    model,
    tools=[search, admin_action],
    prompt="Eres un asistente. Usa las herramientas disponibles.",
    state_schema=AgentState,
    middleware=[logger, AuthMW(), metrics],
)

result = agent.invoke({
    "messages": [("user", "Busca información sobre Python")],
    "user_role": "basic",
})

print(f"\n📊 Métricas: {metrics.calls}")
print(f"📝 Logs: {logger.logs}")
# Output esperado:
# [LOG] MODEL_CALL(2 msgs)
# [LOG] TOOL(search)
# [LOG] MODEL_CALL(4 msgs)
#
# 📊 Métricas: {'model': 2, 'tools': 1}
# 📝 Logs: ['MODEL_CALL(2 msgs)', 'TOOL(search)', 'MODEL_CALL(4 msgs)']

Resumen

En esta cápsula aprendiste:

AgentMiddleware class empaqueta hooks, estado, y tools en una unidad reutilizable — como un plugin para tus agentes
Estructura: defines métodos (before_model, after_model, wrap_model_call, wrap_tool_call, get_tools, get_prompt) dentro de una clase
Composición: pasas middleware=[A(), B(), C()] al agente — LangChain los ejecuta como capas, con A como la más externa
El orden importa: logging/métricas van primero (capturan todo), retry/cache van último (directamente sobre el modelo)
Estado interno: los middleware pueden mantener estado (self.model_calls, self._cache) que persiste entre invocaciones
Parametrizable: como clases Python normales, aceptan __init__ con parámetros para configurar comportamiento (nivel de log, límites de rate, etc.)
Librería organizacional: puedes crear un paquete de middleware (my_middleware/) que todos los equipos importan y componen según sus necesidades
Patrones comunes: logging, auth/permisos, rate limiting, caching, error recovery, métricas — todos se implementan como middleware reutilizables

Próxima cápsula: Proyecto — construirás un agente con routing dinámico de modelos que combina todo lo aprendido en este módulo: middleware compuesto, dynamic models, dynamic tools, y logging.

Recursos adicionales

LangChain Agents Middleware — Guía oficial del sistema de middleware
create_agent API Reference — Referencia con parámetro middleware
LangChain v1.2 Release Notes — Anuncio del middleware system
Middleware Pattern (Martin Fowler) — El patrón de diseño en el que se basa
Python Decorators and Composition — Base conceptual para entender composición de middleware
LangGraph Agents Conceptual Guide — Arquitectura completa de agentes

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

← AnteriorDynamic Tools y Dynamic Prompts Siguiente →Proyecto: Agente con Routing Dinámico de Modelos

AgentMiddleware Class: Middleware Compuesto

Descripción de la cápsula

El problema: hooks individuales no escalan

AgentMiddleware class: la solución

Estructura básica

Crear una middleware class paso a paso

Paso 1: definir el middleware

Paso 2: usar el middleware en un agente

Componer múltiples middleware

El orden importa: cómo se apilan los middleware

Middleware como módulos reutilizables

Middleware parametrizable

Middleware con estado interno

Patrones built-in comunes

Rate limiting middleware

Error recovery middleware

Crear una librería de middleware para tu organización

Estructura recomendada

Ejemplo de módulo compartido

Troubleshooting

1. Los hooks del middleware no se ejecutan

2. wrap_model_call y before_model se ejecutan los dos

3. El middleware con estado se comparte entre invocaciones

4. call_next no está disponible en before_model

5. El orden de middleware no produce el resultado esperado

Ejercicios

Ejercicio 1: Middleware básico de timestamp (Básico)

Ejercicio 2: Middleware de conteo de tokens (Básico)

Ejercicio 3: Componer logging + rate limiting (Intermedio)

Ejercicio 4: Middleware de auditoría con historial (Intermedio)

Ejercicio 5: Middleware que cambia el modelo por costo (Avanzado)

Ejercicio 6: Librería completa con 3 middleware compuestos (Challenge)

Resumen

Recursos adicionales

2. `wrap_model_call` y `before_model` se ejecutan los dos

4. `call_next` no está disponible en `before_model`