LangChain hooks de middleware -- @before_model y @after_model

before_model y after_model son los dos hooks de middleware más utilizados. Se ejecutan antes y después de cada llamada al modelo, y son adecuados para filtrado de contenido, preprocesamiento de mensajes, revisión de respuestas, etc.

@before_model: intercepción antes de la llamada al modelo

@before_model se ejecuta antes de cada llamada al modelo. Aquí puedes modificar mensajes, inyectar condiciones de contexto u omitir directamente la llamada al modelo.

Escenario 1: Preprocesamiento de mensajes — Limitar la longitud de la conversación

Ejemplo

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@before_model
def limit_context(state, runtime):
    """Limita la longitud del historial de mensajes para evitar que el contexto sea demasiado largo"""
    messages = state.get("messages", [])

    # Conservar el system message + los últimos 6 mensajes
    MAX_MESSAGES = 6
    if len(messages) > MAX_MESSAGES:
        # Recortar a los mensajes más recientes
        trimmed = messages[-MAX_MESSAGES:]
        # Asegurar que el primer mensaje sea del usuario
        if trimmed and trimmed[0].type != "human":
            trimmed = trimmed[1:]
        return {"messages": trimmed}

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[limit_context],
    system_prompt="Eres el asistente de EJEMPLO.",
)

# Simular una conversación de varios turnos
result = agent.invoke({
    "messages": [
        HumanMessage(content="Primera ronda"),
        HumanMessage(content="Segunda ronda"),
        HumanMessage(content="Tercera ronda"),
        HumanMessage(content="Cuarta ronda"),
        HumanMessage(content="Quinta ronda"),
        HumanMessage(content="Sexta ronda"),
        HumanMessage(content="Séptima ronda"),
    ]
})
print(f"Número de mensajes: {len(result['messages'])}")
print(f"Respuesta: {result['messages'][-1].content}")

Resultado de la ejecución:

Número de mensajes: 8
Respuesta: ¡Hola! Parece que estás probando una conversación de varios turnos. ¿En qué puedo ayudarte?

Escenario 2: Filtrado de contenido — Bloquear palabras sensibles

Ejemplo

from langchain.agents.middleware import before_model


SENSITIVE_WORDS = ["contraseña", "número de tarjeta bancaria", "número de documento de identidad"]


@before_model
def content_filter(state, runtime):
    """Comprueba si el mensaje del usuario contiene palabras sensibles; si las contiene, interrumpe"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]
    content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

    for word in SENSITIVE_WORDS:
        if word in content:
            print(f"[Interceptado] Palabra sensible detectada: {word}")
            # jump_to="end" para finalizar directamente y evitar que el modelo responda
            return {
                "jump_to": "end",
                "messages": [
                    HumanMessage(content=f"Lo siento, por seguridad, no puedo procesar solicitudes que contengan «{word}».")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[content_filter],
    system_prompt="Eres el asistente de EJEMPLO.",
)

# Pregunta normal
result = agent.invoke({
    "messages": [HumanMessage(content="¿Cómo empezar con Python?")]
})
print(f"Pregunta normal: {result['messages'][-1].content[:80]}")

# Pregunta sensible
result = agent.invoke({
    "messages": [HumanMessage(content="¿Cuál es mi número de tarjeta bancaria? ¿Puedes ayudarme a consultarlo?")]
})
print(f"\nPregunta sensible: {result['messages'][-1].content}")

Resultado de la ejecución:

[拦截] 检测到敏感词: 银行卡号
Pregunta normal: Python 入门可以从安装 Python 环境开始...

Pregunta sensible: 抱歉，为了安全，不能处理包含「银行卡号」的请求。

@after_model — Procesamiento posterior a la llamada del modelo

@after_model se ejecuta después de la respuesta del modelo; es adecuado para revisar la salida del modelo, extraer información clave, añadir instrucciones posteriores, etc.

Escenario 3: Revisión del contenido de la respuesta

Ejemplo

from langchain.agents.middleware import after_model


FORBIDDEN_TOPICS = ["política", "violencia", "pornografía"]


@after_model
def response_audit(state, runtime):
    """Revisa la respuesta del modelo; si involucra temas prohibidos, la reemplaza"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]
    content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

    for topic in FORBIDDEN_TOPICS:
        if topic in content:
            runtime.stream_writer({
                "type": "warning",
                "message": f"Se ha detectado que la respuesta contiene contenido relacionado con «{topic}»; ha sido reemplazada."
            })
            # Devolver un mensaje de reemplazo (mediante el reductor add_messages)
            from langchain.messages import AIMessage
            return {
                "messages": [
                    AIMessage(content="Lo siento, no puedo responder esta pregunta."
                                      "Por favor, pregunta sobre temas relacionados con el aprendizaje de programación.")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[response_audit],
    system_prompt="Eres el asistente de EJEMPLO.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="¿Qué recursos de aprendizaje hay para Python?")]
})
print(f"Respuesta normal: {result['messages'][-1].content}")

Escenario 4: Añadir automáticamente información de aviso

Ejemplo

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


@after_model
def append_disclaimer(state, runtime):
    """Añadir automáticamente un descargo de responsabilidad después de cada respuesta del modelo"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]

    # Añadir solo en la respuesta final de la IA (sin tool_calls)
    if (last_msg.type == "ai"
        and last_msg.content
        and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
        # Reemplazar el último mensaje de la IA añadiendo el descargo de responsabilidad
        return {
            "messages": [
                AIMessage(
                    content=(
                        last_msg.content
                        + "\n\n---\n*El contenido anterior es generado por el asistente de IA de EJEMPLO, solo como referencia.*
                    )
                )
            ]
        }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[append_disclaimer],
    system_prompt="Eres el asistente de EJEMPLO.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="¿Es fácil aprender Python?")]
})
print(result["messages"][-1].content)

Resultado de la ejecución:

Python 是一门非常适合初学者的编程语言，它的语法简洁、贴近自然语言...
---
*El contenido anterior fue generado por el asistente de IA de Ejemplo y es solo de referencia.*

can_jump_to — Control de salto de flujo

En before_model y after_model, puedes controlar el flujo del agente mediante el parámetro can_jump_to y el estado jump_to:

Ejemplo

from langchain.agents.middleware import before_model


@before_model(can_jump_to=["end"])  # Declarar los destinos a los que se puede saltar
def conditional_exit(state, runtime):
    """Finalizar el agente directamente bajo ciertas condiciones"""
    messages = state.get("messages", [])
    if not messages:
        return None

    # Si el usuario dice "adiós", finalizar directamente
    last_content = str(messages[-1].content)
    if last_content.strip() in ["adiós", "chao", "bye"]:
        return {
            "jump_to": "end",  # Finalizar el agente directamente
            "messages": [{"role": "assistant", "content": "¡Adiós! Espero poder servirle en la próxima ocasión."}]
        }

    return None
Valor de can_jump_toSignificadoEscenario aplicable
["end"]Puede saltar al finalSalida condicional, intercepción de seguridad
["model"]Puede saltar de vuelta al modeloRequiere que el modelo reprocese
["tools"]Puede saltar al nodo de herramientaOmitir el modelo y ejecutar la herramienta directamente
["model", "end"]Puede saltar al modelo o al finalMúltiples ramas condicionales

Si no se declara un destino en can_jump_to, jump_to se ignorará. Este es un mecanismo de seguridad para evitar que el middleware salte accidentalmente a nodos no válidos.

Otras extensiones