Fortaleciendo un agente de LangChain en Python con siete middleware integrados
Aprenda cómo el middleware de LangChain 1.0 agrega funcionalidades como resumen, límites de llamadas, reintentos, solución de fallback para modelos, eliminación de información personal sensible y aprobación humana a un agente Gemini sin afectar su lógica principal.
Hacer que un agente de LangChain responda preguntas en una notebook toma solo unos minutos. Sin embargo, conseguir uno en el que se pueda confiar en entornos de producción es más complicado: no debe entrar en bucles infinitos que agoten el presupuesto de la API, ni enviar el número de tarjeta de un cliente al proveedor del modelo, ni remitir correos electrónicos que nadie haya aprobado. LangChain 1.0 aborda estas preocupaciones operativas mediante middleware, una capa de ganchos alrededor del bucle del agente que se configura como una simple lista. Esta guía establece el modelo mental necesario, luego aplica siete middleware a un agente en Python respaldado por Gemini y finaliza con uno personalizado, para que pueda ver exactamente qué cambia cada gancho y cuándo utilizarlo.
Puntos de control alrededor del bucle del agente
Imagínese un aeropuerto. El objetivo es volar de una ciudad a otra, pero alrededor de esta acción principal existen una serie de puntos de control: el check-in confirma la identidad, los escáneres de seguridad revisan el equipaje, la puerta de abordar verifica los pases y la zona de recogida de equipaje se encarga una vez aterrizado. Ninguno de ellos pilota el avión, ni el piloto examina las maletas. Cada capa realiza una tarea antes o después de la acción principal.
El middleware aplica la misma idea a los agentes. La parte central de un agente es un bucle: se llama al modelo, este elige las herramientas, se ejecutan y se repite hasta que el modelo devuelve una respuesta final. El middleware inserta puntos de control alrededor de ese bucle sin modificarlo:
- Eliminar los números de tarjeta antes de que el texto llegue al LLM es lo que hace el escáner de seguridad.
- Requerir que una persona apruebe un correo electrónico enviado es lo que corresponde a la puerta de abordar.
- Detenerse después de diez llamadas al modelo para limitar los gastos es lo que hace el disyuntor de circuito.
Si trabajas con TypeScript, la guía complementaria en LangChain guardrails and middleware aborda las mismas ideas desde el lado de JavaScript; esta guía se centra en Python y se enfoca en las clases integradas concretas.
Los ganchos que puede utilizar el middleware
El bucle expone ganchos en cada etapa, y un middleware se conecta a uno o más de ellos:
before_agentyafter_agentse ejecutan una vez, al principio y al final de cada llamada.before_modelyafter_modelse activan cada vez que el bucle está a punto de llamar al modelo o acaba de hacerlo.wrap_model_callywrap_tool_callenvuelven la llamada en sí, lo que permite reintentarla, sustituirla o obtenerla desde la caché.
Ese es todo el modelo mental. Se añaden los middleware pasando una lista a create_agent, como en este ejemplo que combina la redacción de correos electrónicos, un límite de llamadas y reintentos de herramientas:
agent = create_agent(
model=model,
tools=[my_tool],
middleware=[
PIIMiddleware("email", strategy="redact"),
ModelCallLimitMiddleware(run_limit=5),
ToolRetryMiddleware(max_retries=3),
],
)
El orden de la lista es importante. Los middleware se aplican en secuencia, como las capas de una cebolla que rodean al agente, por lo que el paso de redacción listado primero procesa la entrada bruta antes que cualquier otro.
Configuración y un agente básico
Los middleware requieren LangChain 1.0 o una versión posterior, así que instálalos con la flag -U para actualizar cualquier versión anterior. Los ejemplos utilizan Gemini a través de su nivel gratuito; puedes crear una clave en Google AI Studio.
!pip install -qU langchain langchain-google-genai
Importa os y la clase del modelo de chat Gemini:
import os
from langchain_google_genai import ChatGoogleGenerativeAI
Luego se configura el modelo. El fragmento muestra la clave directamente como ejemplo; en código real, se debe exportar GOOGLE_API_KEY en el entorno o cargarla desde un gestor de secretos en lugar de incluirla en el código fuente. Una temperatura de cero permite que los resultados sean reproducibles:
os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)
El sujeto de cada experimento es un agente mínimo sin middleware. Necesita la función create_agent y el decorador tool:
from langchain.agents import create_agent
from langchain_core.tools import tool
Cuenta con una herramienta meteorológica ficticia que siempre indica sol, y se invoca con un único mensaje del usuario:
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)
Cada sección a continuación añade otro punto de control alrededor de este agente.
1. SummarizationMiddleware para memoria limitada
En conversaciones largas, el historial de mensajes sigue creciendo hasta superar la ventana de contexto, y cada token adicional se cobra. SummarizationMiddleware verifica el tamaño del historial en before_model; cuando supera un umbral, condensa los mensajes más antiguos en un resumen y mantiene solo los más recientes tal como están.
from langchain.agents.middleware import SummarizationMiddleware
La configuración a continuación utiliza el mismo modelo para escribir resúmenes, se activa al llegar a 10 mensajes y mantiene los 4 más recientes sin modificarlos. La prueba crea un historial falso de seis ciudades (doce mensajes) y luego pregunta cuál ciudad apareció primero:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
SummarizationMiddleware(
model=model, # which LLM writes the summary
trigger=("messages", 10), # summarize when history hits 10 messages
keep=("messages", 4), # keep the 4 most recent messages intact
),
],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history)) # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"])) # 6
Entran trece mensajes y quedan seis después: el resumen, los cuatro mensajes conservados y la nueva respuesta. El modelo sigue respondiendo “Delhi”, porque ese dato se incluyó en el resumen. Contar los mensajes permite observar fácilmente este comportamiento en una demostración, pero en producción un disparador basado en tokens como (“tokens”, 3000) o una fracción de la ventana de contexto como (“fraction”, 0.8) permiten controlar mucho mejor el costo real y establecer límites más adecuados. Tenga en cuenta que los resúmenes causan pérdida de información: las cifras exactas o identificadores mencionados al principio de la conversación pueden no sobrevivir.
2. Límites de llamadas como interruptores de costo
El fallo del agente más costoso es el bucle infinito, en el que el modelo y las herramientas siguen llamándose mutuamente y generando gastos durante minutos antes de que alguien se dé cuenta. Dos middleware imponen un techo estricto a esto:
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
Aquí el modelo tiene un límite de tres llamadas por ejecución, con exit_behavior="end" para que el agente se detenga de forma ordenada en lugar de generar una excepción, y las herramientas tienen un límite de dos llamadas. La instrucción solicita deliberadamente seis ciudades, una por una:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
# "end" = stop gracefully instead of raising an error
ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
ToolCallLimitMiddleware(run_limit=2),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)
El agente necesita realizar seis búsquedas, alcanza sus límites y finaliza de manera adecuada con los resultados parciales que ha obtenido. También existe un parámetro thread_limit para limitar las llamadas en todo el hilo de conversación en lugar de solo en una ejecución. Estos dos mecanismos son una protección básica; elija límites suficientemente altos por encima de lo que requieren las solicitudes legítimas para que se activen únicamente en casos reales de descontrol.
3. ToolRetryMiddleware para dependencias inestables
Las herramientas reales fallan: las llamadas HTTP se agotan y las conexiones se interrumpen. ToolRetryMiddleware utiliza wrap_tool_call para detectar los fallos y volver a intentarlos con un retroceso exponencial.
from langchain.agents.middleware import ToolRetryMiddleware
Para demostrarlo, una herramienta que calcula precios de acciones cuenta sus intentos y genera un ConnectionError en los dos primeros antes de tener éxito. El middleware permite hasta tres intentos, comenzando con un retraso de un segundo y duplicándolo cada vez:
attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
"""Get the current stock price for a ticker symbol."""
attempt_counter["count"] += 1
print(f" [tool called — attempt #{attempt_counter['count']}]")
if attempt_counter["count"] < 3:
raise ConnectionError("API timeout — please retry")
return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
model=model,
tools=[flaky_stock_price],
middleware=[
ToolRetryMiddleware(
max_retries=3, # retry a failed tool up to 3 times
initial_delay=1.0, # wait 1s before first retry
backoff_factor=2.0, # double the wait each time: 1s, 2s, 4s
),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)
La herramienta falla dos veces, el middleware espera y vuelve a intentarlo, y el tercer intento tiene éxito. Desde la perspectiva del agente, nada salió mal. Las reintentos solo son seguros para operaciones idempotentes como las lecturas; volver a intentar una herramienta que carga una tarjeta o envía un mensaje podría repetir el efecto secundario.
4. ModelFallbackMiddleware para interrupciones del proveedor
La misma idea de resiliencia puede proteger las llamadas al modelo. Cuando el modelo principal falla, por ejemplo debido a limitaciones de velocidad o una interrupción del servicio, este middleware vuelve a intentar la solicitud con modelos de respaldo en el orden en que se hayan listado:
from langchain.agents.middleware import ModelFallbackMiddleware
En el ejemplo, se utiliza el modelo Gemini más ligero como principal y se agrega un segundo modelo Gemini como respaldo:
backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
model=model, # primary: gemini-3.5-flash-lite
tools=[get_weather],
middleware=[ModelFallbackMiddleware(backup_model)],
)
Mientras el modelo principal esté funcionando correctamente, no se notará nada, que es precisamente el objetivo. El valor solo se hace evidente el día en que su proveedor tenga problemas. Para una protección más sólida, considere utilizar un modelo de respaldo de otro proveedor, ya que las interrupciones suelen afectar a todos los modelos detrás de la misma API.
5. PIIMiddleware para datos sensibles
Frecuentemente no se desea que los correos electrónicos, números de tarjeta o direcciones IP sean enviados en absoluto al proveedor del modelo. PIIMiddleware escanea el texto en before_model, antes de que el modelo lo vea, y aplica una de cuatro estrategias: redact, mask, hash o block.
from langchain.agents.middleware import PIIMiddleware
Este agente no cuenta con herramientas adicionales. Reemplaza completamente las direcciones de correo electrónico por un marcador de posición y enmascara los números de tarjeta de modo que solo queden los cuatro últimos dígitos; ambas reglas se aplican a la entrada del usuario:
agent = create_agent(
model=model,
tools=[],
middleware=[
# Replace emails entirely with [REDACTED_EMAIL]
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# Mask credit cards — keeps last 4 digits
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Draft a support reply to priya.sharma@example.com confirming her card "
"4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)
El modelo redacta su respuesta sin recibir nunca la dirección real ni el número completo de la tarjeta. También se puede registrar un tipo personalizado de PII con una expresión regular propia, por ejemplo, para bloquear cualquier texto que parezca ser una clave de API interna:
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
La detección basada en patrones captura valores bien formados, no todas las formas creativas de escribir, por lo que debe considerarse como una capa de defensa y no como una garantía de cumplimiento.
6. HumanInTheLoopMiddleware para puertas de aprobación
Algunas acciones son demasiado importantes como para permitir autonomía total: enviar correos electrónicos, eliminar registros o realizar pagos. HumanInTheLoopMiddleware detiene al agente inmediatamente antes de que se ejecute una herramienta sensible, espera la decisión de una persona y luego continúa.
Esto funciona de manera diferente a los middleware anteriores. El agente pausado no se termina. Un checkpointer guarda su estado completo, y el ID del hilo sirve como clave para encontrarlo y reanudar su ejecución más tarde. El revisor puede aprobar la llamada, editar sus argumentos o rechazarla.
Las importaciones incluyen el middleware, un controlador en memoria de LangGraph y el tipo Command utilizado para reanudar:
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
El agente a continuación cuenta con una herramienta send_email, la marca como interrumpible y almacena el estado pausado en InMemorySaver. La primera llamada, en el hilo demo-1, le pide al agente que envíe un correo electrónico a un administrador:
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
model=model,
tools=[send_email],
middleware=[
# Pause and ask a human whenever the agent wants to call send_email
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
],
checkpointer=InMemorySaver(), # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
{"messages": [{"role": "user", "content":
"Send an email to boss@company.com saying the report is ready."}]},
config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])
La ejecución se detiene antes de que la herramienta se ejecute. La entrada __interrupt__ en el resultado muestra la llamada pendiente con su destinatario, asunto y cuerpo, y no se ha enviado nada. La aprobación se envía como un Command en el mismo hilo, lo que reanuda la ejecución pausada:
# Step 2: approve and resume
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config, # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)
En lugar de approve, puedes enviar reject con una razón o edit con argumentos modificados. En una aplicación real, es en este punto donde se mostraría una pantalla de aprobación. Dos observaciones prácticas: InMemorySaver pierde el estado cuando se reinicia el proceso, por lo que los sistemas en producción necesitan un punto de control persistente; además, el formato del payload para reanudar ha cambiado entre las versiones de LangChain, así que confirma su formato en la referencia de middleware correspondiente a la versión que estás utilizando.
7. Creando tu propio middleware con un decorador
Cuando nada de lo integrado sirve, un middleware personalizado resulta sencillo de crear, ya que cada gancho cuenta con su propio decorador correspondiente. Las importaciones necesarias son el decorador before_model y el tipo AgentState:
from langchain.agents.middleware import before_model, AgentState
Este ejemplo registra cuántos mensajes están a punto de ser enviados en cada llamada al modelo y se agrega a la lista como cualquier middleware preconstruido:
@before_model
def log_before_model(state: AgentState, runtime) -> None:
print(f" [middleware] Calling model with {len(state['messages'])} messages")
# Returning None = observe only.
# Returning a dict would UPDATE the agent's state (e.g., trim messages)
return Noneagent = create_agent(
model=model,
tools=[get_weather],
middleware=[log_before_model], # plugs in like any prebuilt middleware
)
El valor de retorno es la elección de diseño importante. Devolver None significa que el middleware solo observa. Devolver un diccionario actualiza el estado del agente, y así es como se pueden filtrar mensajes, injectar contexto o aplicar restricciones personalizadas. También existen decoradores para los otros puntos de conexión: @before_agent, @after_model, @wrap_model_call, @wrap_tool_call, además de @dynamic_prompt para crear prompts del sistema en tiempo de ejecución.
Puntos clave
- El middleware separa las preocupaciones operativas de la lógica del agente: el bucle permanece igual mientras se añaden puntos de control alrededor de él como una lista simple pasada a
create_agent.
TodoListMiddleware, LLMToolSelectorMiddleware y ContextEditingMiddleware, está documentado en la referencia oficial.Lecturas relacionadas
- De un chatbot de un nodo a un agente respaldado por MCP en LangGraph — Construye una aplicación de LangGraph capa por capa: estado y reductores, aristas, bucles de herramientas, hilos con puntos de control, tres modos de transmisión en tiempo real y herramientas proporcionadas a través de MCP.
- Memoria de agente híbrido: Fusionando BM25 y búsqueda vectorial con RRF en Python — Aprende por qué la búsqueda vectorial pura falla como memoria de agente, cómo la fusión de rango recíproco combina los resultados de BM25 y los densos en Python, y cuándo las resúmenes de GraphRAG son útiles.