Hooks de middleware LangChain -- @before_model et @after_model

before_model et after_model sont les deux hooks de middleware les plus couramment utilisés. Ils s'exécutent avant et après chaque appel de modèle, adaptés au filtrage de contenu, au prétraitement des messages, à la révision des réponses, etc.

@before_model — Interception avant l'appel du modèle

@before_model s'exécute avant chaque appel de modèle. Ici, vous pouvez modifier les messages, injecter des conditions de contexte ou ignorer complètement l'appel du modèle.

Scénario 1 : Prétraitement des messages — Limitation de la longueur de la conversation

Exemple

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):
    """Limiter la longueur de l'historique des messages pour éviter un contexte excessivement long"""
    messages = state.get("messages", [])

    # Conserver le message système + les 6 messages les plus récents
    MAX_MESSAGES = 6
    if len(messages) > MAX_MESSAGES:
        # Ne garder que les messages les plus récents
        trimmed = messages[-MAX_MESSAGES:]
        # S'assurer que le premier message est un message utilisateur
        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="Vous êtes l'assistant du tutoriel EXAMPLE.",
)

# Simuler une conversation multi-tours
result = agent.invoke({
    "messages": [
        HumanMessage(content="Tour 1"),
        HumanMessage(content="Tour 2"),
        HumanMessage(content="Tour 3"),
        HumanMessage(content="Tour 4"),
        HumanMessage(content="Tour 5"),
        HumanMessage(content="Tour 6"),
        HumanMessage(content="Tour 7"),
    ]
})
print(f"Nombre de messages : {len(result['messages'])}")
print(f"Réponse : {result['messages'][-1].content}")

Sortie :

Message count: 8
Reply: Hello! It looks like you are testing multi-turn conversation. How can I help you?

Scénario 2 : Filtrage de contenu — Blocage des mots sensibles

Exemple

from langchain.agents.middleware import before_model


SENSITIVE_WORDS = ["mot de passe", "numéro de carte bancaire", "numéro de carte d'identité"]


@before_model
def content_filter(state, runtime):
    """Vérifie si les messages utilisateur contiennent des mots sensibles et les intercepte si c'est le cas"""
    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"[Bloqué] Mot sensible détecté : {word}")
            # jump_to="end" termine directement, empêchant le modèle de répondre
            return {
                "jump_to": "end",
                "messages": [
                    HumanMessage(content=f"Désolé, pour des raisons de sécurité, les requêtes contenant '{word}' ne peuvent pas être traitées.")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[content_filter],
    system_prompt="Vous êtes l'assistant du tutoriel EXAMPLE.",
)

# Question normale
result = agent.invoke({
    "messages": [HumanMessage(content="Comment démarrer avec Python ?")]
})
print(f"Question normale : {result['messages'][-1].content[:80]}")

# Question sensible
result = agent.invoke({
    "messages": [HumanMessage(content="Quel est le numéro de ma carte bancaire ? Pouvez-vous m'aider à le vérifier ?")]
})
print(f"\nQuestion sensible : {result['messages'][-1].content}")

Sortie :

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

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

@after_model — Traitement après l'appel du modèle

@after_model s'exécute après la réponse du modèle, ce qui convient pour examiner la sortie du modèle, extraire des informations clés, ajouter des instructions de suivi, etc.

Scénario 3 : examen du contenu de la réponse

Exemple

from langchain.agents.middleware import after_model


FORBIDDEN_TOPICS = ["politique", "violence", "pornographie"]


@after_model
def response_audit(state, runtime):
    """Examine les réponses du modèle et remplace-les si elles impliquent des sujets interdits"""
    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"Réponse détectée comme contenant du contenu lié à '{topic}', a été remplacée"
            })
            # Retourner un message de remplacement (via add_messages reducer)
            from langchain.messages import AIMessage
            return {
                "messages": [
                    AIMessage(content="Désolé, je ne peux pas répondre à cette question."
                                      "Veuillez poser des questions sur des contenus liés à l'apprentissage de la programmation.")
                ]
            }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[response_audit],
    system_prompt="Vous êtes l'assistant du tutoriel EXAMPLE.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Quelles ressources d'apprentissage sont disponibles pour Python ?")]
})
print(f"Réponse normale : {result['messages'][-1].content}")

Scénario 4 : Ajout automatique des informations de prompt

Exemple

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


@after_model
def append_disclaimer(state, runtime):
    """Ajoute automatiquement une clause de non-responsabilité après chaque réponse du modèle"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]

    # N'ajouter que lorsqu'il s'agit de la réponse finale de l'IA (pas de tool_calls)
    if (last_msg.type == "ai"
        and last_msg.content
        and not (hasattr(last_msg, 'tool_calls') and last_msg.tool_calls)):
        # Remplacer le dernier message de l'IA, en ajoutant l'avertissement
        return {
            "messages": [
                AIMessage(
                    content=(
                        last_msg.content
                        + "\n\n---\n*Le contenu ci-dessus a été généré par l'assistant IA EXAMPLE et est fourni à titre de référence uniquement.*
                    )
                )
            ]
        }

    return None


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[append_disclaimer],
    system_prompt="Vous êtes l'assistant d'EXAMPLE.",
)

result = agent.invoke({
    "messages": [HumanMessage(content="Python est-il facile à apprendre ?")]
})
print(result["messages"][-1].content)

Sortie :

Python 是一门非常适合初学者的编程语言，它的语法简洁、贴近自然语言...
---
*The above content was generated by the Example AI assistant and is for reference only.*

can_jump_to — Contrôle de saut de flux

Dans before_model et after_model, vous pouvez contrôler le flux de l'Agent via le paramètre can_jump_to et l'état jump_to :

Exemple

from langchain.agents.middleware import before_model


@before_model(can_jump_to=["end"])  # Déclarer les cibles vers lesquelles il est possible de sauter
def conditional_exit(state, runtime):
    """Terminer directement l'Agent dans des conditions spécifiques"""
    messages = state.get("messages", [])
    if not messages:
        return None

    # Si l'utilisateur dit "goodbye", terminer directement
    last_content = str(messages[-1].content)
    if last_content.strip() in ["Goodbye", "Bye", "bye"]:
        return {
            "jump_to": "end",  # Terminer directement l'Agent
            "messages": [{"role": "assistant", "content": "Au revoir ! Au plaisir de vous servir la prochaine fois."}]
        }

    return None
Valeur de can_jump_toSignificationScénarios applicables
["end"]Peut sauter vers la finSortie conditionnelle, interception de sécurité
["model"]Peut revenir au modèleLorsque le modèle doit retraiter
["tools"]Peut sauter vers le nœud d'outilIgnorer le modèle et exécuter l'outil directement
["model", "end"]Peut sauter vers le modèle ou la finBranches conditionnelles multiples

Si la cible n'est pas déclarée dans can_jump_to, jump_to sera ignoré. Il s'agit d'un mécanisme de sécurité destiné à empêcher le middleware de sauter accidentellement vers des nœuds illégaux.

Autres extensions