LangChain Middleware
LangChain Middleware (middleware) es la característica más potente de LangChain. Te permite insertar lógica personalizada en cada etapa de la ejecución del Agent, implementando reintentos, degradación, caché, filtrado de contenido, registro de logs, etc., sin necesidad de modificar el código del propio Agent.
Qué es Middleware
Middleware es elhook (Hook). Cada hook te permite ejecutar código personalizado en un punto específico del flujo de ejecución:
Ejemplo
# Comprensión intuitiva de Middleware:
# Supongamos que el flujo de ejecución del Agent es así:
# 1. Entrada del usuario → 2. Pensamiento del modelo → 3. Posible llamada a herramientas → 4. El modelo vuelve a pensar → 5. Salida del resultado
# Middleware te permite insertar lógica personalizada entre estos 5 pasos:
# 1. Entrada del usuario
# ↓ [hook before_agent: registro de logs, comprobación de permisos]
# 2. Pensamiento del modelo
# ↓ [hook before_model: preprocesamiento de mensajes]
# ↓ [hook wrap_model_call: reintento, degradación, caché]
# ↓ [hook after_model: revisión de contenido]
# 3. Ejecución de herramientas
# ↓ [hook wrap_tool_call: reintento de llamada a herramientas]
# 4. Volver al pensamiento del modelo (bucle hasta completar)
# ↓ [hook after_agent: formato de resultados, análisis estadístico]
# 5. Salida del resultado
# Supongamos que el flujo de ejecución del Agent es así:
# 1. Entrada del usuario → 2. Pensamiento del modelo → 3. Posible llamada a herramientas → 4. El modelo vuelve a pensar → 5. Salida del resultado
# Middleware te permite insertar lógica personalizada entre estos 5 pasos:
# 1. Entrada del usuario
# ↓ [hook before_agent: registro de logs, comprobación de permisos]
# 2. Pensamiento del modelo
# ↓ [hook before_model: preprocesamiento de mensajes]
# ↓ [hook wrap_model_call: reintento, degradación, caché]
# ↓ [hook after_model: revisión de contenido]
# 3. Ejecución de herramientas
# ↓ [hook wrap_tool_call: reintento de llamada a herramientas]
# 4. Volver al pensamiento del modelo (bucle hasta completar)
# ↓ [hook after_agent: formato de resultados, análisis estadístico]
# 5. Salida del resultado
Seis puntos de hook
El Middleware de LangChain proporciona 6 hooks, divididos en dos categorías según el momento de ejecución:
| Hook | Frecuencia de ejecución | Ubicación de ejecución | Uso principal |
|---|---|---|---|
| before_agent | Una vez | Antes de que comience el Agent | Inicialización, comprobación de permisos, preprocesamiento de entrada |
| before_model | Cada ciclo | Antes de la llamada al modelo | Preprocesamiento de mensajes, inyección dinámica de contexto |
| wrap_model_call | Cada ciclo | Envolver la llamada al modelo | Reintentos, degradación, caché, reescritura de solicitudes |
| after_model | Cada ciclo | Después de la llamada al modelo | Revisión de contenido, filtrado de respuestas, logs |
| wrap_tool_call | Cada llamada a herramienta | Envolver la ejecución de herramientas | Reintentos de herramientas, caché de resultados, reescritura de parámetros |
| after_agent | Una vez | Después de que el Agent finalice | Formato de salida, estadísticas, limpieza de recursos |
Dos formas de uso
Middleware se puede utilizar de dos formas: mediante herencia de clases o mediante decoradores.
Forma 1: Decorador (recomendado)
Ejemplo
from langchain.agents.middleware import before_model, after_model
# Forma de decorador: simple e intuitiva
@before_model
def log_before(state, runtime):
"""Registrar un log antes de cada llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Número de mensajes actual: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Registrar un log después de cada llamada al modelo"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] El modelo solicitó una llamada a herramienta")
return None
# Forma de decorador: simple e intuitiva
@before_model
def log_before(state, runtime):
"""Registrar un log antes de cada llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Número de mensajes actual: {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Registrar un log después de cada llamada al modelo"""
last_msg = state["messages"][-1] if state.get("messages") else None
if last_msg and hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
print(f"[after_model] El modelo solicitó una llamada a herramienta")
return None
Forma 2: Herencia de clases (adecuado para lógica compleja)
Ejemplo
from langchain.agents.middleware import AgentMiddleware
class LoggingMiddleware(AgentMiddleware):
"""Middleware de registro de logs personalizado"""
@property
def name(self) -> str:
# Nombre personalizado del middleware (por defecto es el nombre de la clase)
return "logging"
def before_agent(self, state, runtime):
"""Lógica antes de que comience el Agent"""
print("[Logging] El Agent comienza a ejecutarse")
return None
def before_model(self, state, runtime):
"""Lógica antes de la llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f"[Logging] Preparándose para llamar al modelo, actualmente {msg_count} mensajes")
return None
def after_model(self, state, runtime):
"""Lógica después de la llamada al modelo"""
print("[Logging] Llamada al modelo completada")
return None
def after_agent(self, state, runtime):
"""Lógica después de que el Agent finalice"""
print("[Logging] Ejecución del Agent finalizada")
return None
class LoggingMiddleware(AgentMiddleware):
"""Middleware de registro de logs personalizado"""
@property
def name(self) -> str:
# Nombre personalizado del middleware (por defecto es el nombre de la clase)
return "logging"
def before_agent(self, state, runtime):
"""Lógica antes de que comience el Agent"""
print("[Logging] El Agent comienza a ejecutarse")
return None
def before_model(self, state, runtime):
"""Lógica antes de la llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f"[Logging] Preparándose para llamar al modelo, actualmente {msg_count} mensajes")
return None
def after_model(self, state, runtime):
"""Lógica después de la llamada al modelo"""
print("[Logging] Llamada al modelo completada")
return None
def after_agent(self, state, runtime):
"""Lógica después de que el Agent finalice"""
print("[Logging] Ejecución del Agent finalizada")
return None
Ejemplo completo del ciclo de vida
Ejemplo
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Antes de que comience el Agent"""
print(">>> [before_agent] El Agent comienza <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Antes de cada llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Mensaje {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Después de cada llamada al modelo"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Herramienta solicitada: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Respuesta directa: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Después de que el Agent finalice"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] El Agent finalizó, {total_msgs} mensajes en total <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Consultar el clima"""
return f"{city}: Despejado, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Eres un asistente.",
)
print("\n"========== Primera pregunta (requiere herramienta) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="¿Clima en Hangzhou?")]
})
print(f"\n"Respuesta final: {result['messages'][-1].content}")
print("\n"========== Segunda pregunta (sin necesidad de herramienta) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hola")]
})
print(f"\n"Respuesta final: {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_agent, after_agent,
before_model, after_model,
)
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@before_agent
def start_log(state, runtime):
"""Antes de que comience el Agent"""
print(">>> [before_agent] El Agent comienza <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Antes de cada llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Mensaje {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Después de cada llamada al modelo"""
last = state["messages"][-1] if state.get("messages") else None
if hasattr(last, 'tool_calls') and last.tool_calls:
tools = [tc['name'] for tc in last.tool_calls]
print(f" <- [after_model] Herramienta solicitada: {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Respuesta directa: {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Después de que el Agent finalice"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] El Agent finalizó, {total_msgs} mensajes en total <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Consultar el clima"""
return f"{city}: Despejado, 25°C"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[start_log, pre_model, post_model, end_log],
system_prompt="Eres un asistente.",
)
print("\n"========== Primera pregunta (requiere herramienta) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="¿Clima en Hangzhou?")]
})
print(f"\n"Respuesta final: {result['messages'][-1].content}")
print("\n"========== Segunda pregunta (sin necesidad de herramienta) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hola")]
})
print(f"\n"Respuesta final: {result['messages'][-1].content}")
Resultado de la ejecución:
========== 第一个问题（需要工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 2 条消息 <- [after_model] 请求工具: ['get_weather'] -> [before_model] 第 3 条消息 <- [after_model] 直接回复: 杭州今天晴，气温25°C。... <<< [after_agent] Agent 结束，共 4 条消息 <<< Respuesta final: 杭州今天晴，气温25°C. ========== 第二个问题（无需工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 1 条消息 <- [after_model] 直接回复: 你好！有什么可以帮你的？... <<< [after_agent] Agent 结束，共 2 条消息 <<< Respuesta final: ¡Hola! ¿En qué puedo ayudarte?
En la salida se puede ver:
- before_agent y after_agent: solo se ejecutan una vez por pregunta
- before_model y after_model: se ejecutan en cada llamada al modelo (la primera pregunta llamó al modelo dos veces, por lo que cada uno se ejecutó dos veces)
El valor de retorno del Middleware
El valor de retorno del Middleware determina si se debe modificar el estado del Agent o controlar el flujo:
| Valor de retorno | Efecto | Ejemplo |
|---|---|---|
| None | No modifica ningún estado, continúa con el flujo normal | Solo registro de logs |
| dict | Actualiza el estado del Agent (se fusiona con el estado actual) | Devuelve {"custom_field": "value"} |
| Dict que contiene jump_to | Salta al nodo especificado | Devuelve {"jump_to": "end"} |
Otras extensionesEl dict devuelto se fusiona a través del reducer del estado del Agent. Para el campo messages, se utiliza el reducer add_messages, por lo que los messages devueltos se agregan en lugar de sobrescribirse.