LangChain @before_agent y @after_agent

before_agent y after_agent son hooks a nivel de Agent, que se ejecutan una vez antes de la ejecución del Agent y una vez después de su finalización, respectivamente. Son adecuados para inicialización, preprocesamiento, postprocesamiento y análisis estadístico.

before_agent -- Trabajos preparatorios antes de que comience el Agent

before_agent se ejecuta antes de que el Agent comience formalmente su ejecución, y solo se ejecuta una vez. Aquí puedes realizar preprocesamiento de entrada, validación de información del usuario, inicialización de recursos, etc.

Escenario 1: Preprocesamiento de entrada -- Corrección automática de la entrada del usuario

Ejemplo

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import before_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

@before_agent
def preprocess_input(state, runtime):
"""Procesar la entrada del usuario antes de que comience el Agent"""
messages = state.get("messages", [])
if not messages:
return None

# Obtener el último mensaje del usuario
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

# Añadir automáticamente lenguaje cortés (si el usuario pregunta directamente)
greetings = ["Hola", "Saludos", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# No modificar, devolver directamente
pass

return None

@tool
def search_course(keyword: str) -> str:
"""Buscar cursos en Ejemplo"""
courses = {
"python": "Tutorial básico de Python3 (gratuito, 30 capítulos)",
"html": "Tutorial básico de HTML (gratuito, 25 capítulos)",
}
return courses.get(keyword.lower(), f"No se encontraron cursos para {keyword}")

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="Eres el asesor de cursos de Ejemplo.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Cursos de Python")]
})
print(f"Respuesta: {result['messages'][-1].content}")

Escenario 2: Control de acceso -- Verificación de permisos

Ejemplo

from langchain.agents.middleware import before_agent

@before_agent
def access_control(state, runtime):
"""Verificar si el usuario tiene permiso para usar el Agent"""
# Obtener información del usuario desde runtime.context
context = runtime.context
if context is None:
return None

user_role = context.get("user_role", "guest")

# Los usuarios invitados solo pueden usar funciones limitadas
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Comprobar si se trata de funciones restringidas
restricted_keywords = ["eliminar", "administrar", "configurar", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="Su permiso actual es insuficiente para realizar esta operación. Inicie sesión e inténtelo de nuevo."
)]
}

return None

after_agent -- Procesamiento posterior a la finalización del Agent

after_agent se ejecuta después de que el Agent completa todo el procesamiento (solo una vez). Aquí puedes formatear la salida final, registrar información estadística, limpiar recursos, etc.

Escenario 3: Análisis estadístico -- Registro de datos de conversación

Ejemplo

from langchain.agents.middleware import after_agent

@after_agent
def conversation_stats(state, runtime):
"""Contar la información de la conversación y añadirla al resultado"""
messages = state.get("messages", [])

# Datos estadísticos
model_calls = 0
tool_calls = 0
total_chars = 0

for msg in messages:
if msg.type == "ai":
model_calls += 1
if hasattr(msg, 'tool_calls') and msg.tool_calls:
tool_calls += len(msg.tool_calls)
if hasattr(msg, 'content') and msg.content:
total_chars += len(str(msg.content))

# Enviar información estadística a través de custom stream
runtime.stream_writer({
"type": "stats",
"model_calls": model_calls,
"tool_calls": tool_calls,
"total_messages": len(messages),
"total_chars": total_chars,
})

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[conversation_stats],
system_prompt="Eres el asesor de cursos de Ejemplo.",
)

# Usar stream_mode=["updates", "custom"] para recibir eventos personalizados
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Consulta los cursos de Python")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Información estadística: {chunk}")

Resultado de la ejecución:

统计信息: {'type': 'stats', 'model_calls': 2, 'tool_calls': 1,
            'total_messages': 4, 'total_chars': 127}

Escenario 4: Formateo de salida -- Unificación del estilo de respuesta

Ejemplo

from langchain.agents.middleware import after_agent
from langchain.messages import AIMessage

@after_agent
def format_output(state, runtime):
"""Añadir información de resumen formateada al resultado"""
messages = state.get("messages", [])
if not messages:
return None

# Encontrar el último mensaje de IA (respuesta final)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break

if last_ai:
# Información estadística
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)

footer = (
f"\n\n---\n"
f"> Esta conversación consta de {len(messages)} mensajes en total,
f"se llamaron las herramientas {tool_count} veces.\n"
f"> Con el apoyo del asistente de IA de Ejemplo."
)

# Añadir a la respuesta final
return {
"messages": [
AIMessage(content=last_ai.content + footer)
]
}

return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[format_output],
system_prompt="Eres el asesor de cursos de Ejemplo.",
)

result = agent.invoke({
"messages": [HumanMessage(content="¿Qué cursos hay de Python?")]
})
print(result["messages"][-1].content)

Resultado de la ejecución:

Ejemplo ofrece el tutorial básico de Python3 en 30 capítulos, totalmente gratuito y perfecto para principiantes de Python.

---
> 本次对话共进行 4 条消息，调用了 1 次工具。
> 由Ejemplo AI 助手提供支持。

Ejemplo completo de colaboración de los cuatro hooks

Ejemplo

from langchain.agents.middleware import (
before_agent, after_agent, before_model, after_model
)
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool

# ----- Definir todos los hooks -----

@before_agent
def init_session(state, runtime):
"""Inicio: Inicializar la sesión"""
print(">>> Inicio de la sesión")
return None

@before_model
def pre_model_check(state, runtime):
"""Antes de cada llamada al modelo"""
msg_count = len(state.get("messages", []))
print(f" [model antes] Número de mensajes: {msg_count}")
return None

@after_model
def post_model_check(state, runtime):
"""Después de cada llamada al modelo"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [model después] Se necesita llamada a herramienta")
return None

@after_agent
def finish_session(state, runtime):
"""Fin: Limpiar recursos"""
total = len(state.get("messages", []))
print(f"<<< Fin de la sesión, {total} mensajes en total")
return None

# ----- Crear Agent -----

@tool
def get_weather(city: str) -> str:
"""Consultar el clima"""
return f"{city}: Despejado"

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[init_session, pre_model_check, post_model_check, finish_session],
system_prompt="Eres un asistente.",
)

result = agent.invoke({
"messages": [HumanMessage(content="¿Clima en Hangzhou?")]
})
print(f"\nRespuesta final: {result['messages'][-1].content}")

Resultado de la ejecución:

>>> 会话开始
  [model前] 消息数: 2
  [model后] 需要工具调用
  [model前] 消息数: 3
<<< 会话结束，共 4 条消息

Respuesta final: Hoy hace sol en Hangzhou, ideal para salir.

Resumen de los hooks de Middleware

Hook	Número de ejecuciones	Cuándo usar	Capacidades clave
before_agent	1 vez	Verificación de permisos, preprocesamiento de entrada, inicialización de recursos	Puede usar jump_to="end" para terminar antes
before_model	Cada ciclo	Recorte de mensajes, filtrado de contenido, inyección de contexto	Puede usar jump_to para controlar el flujo
wrap_model_call	Cada ciclo	Reintento, degradación, caché, modificación de prompt	Control total de la ejecución del modelo
after_model	Cada ciclo	Revisión de respuestas, adición de contenido, registros	Puede reemplazar la salida del modelo
wrap_tool_call	Cada llamada a herramienta	Reintento de herramientas, caché, reescritura de parámetros	Control total de la ejecución de herramientas
after_agent	1 vez	Formateo de salida, análisis estadístico, limpieza	Modificación del estado final

Otras extensiones

Agent Configuración de desarrollo

Manual de referencia

LangChain @before_agent y @after_agent

before_agent -- Trabajos preparatorios antes de que comience el Agent

Escenario 1: Preprocesamiento de entrada -- Corrección automática de la entrada del usuario

Ejemplo

Escenario 2: Control de acceso -- Verificación de permisos

Ejemplo

after_agent -- Procesamiento posterior a la finalización del Agent

Escenario 3: Análisis estadístico -- Registro de datos de conversación

Ejemplo

Escenario 4: Formateo de salida -- Unificación del estilo de respuesta

Ejemplo

Ejemplo completo de colaboración de los cuatro hooks

Ejemplo

Resumen de los hooks de Middleware