LangChain @before_agent et @after_agent

before_agent et after_agent sont des hooks au niveau de l'Agent qui s'exécutent une fois respectivement avant et après l'exécution de l'Agent. Ils conviennent à l'initialisation, au prétraitement, au post-traitement et à l'analyse statistique.

before_agent -- Préparation avant le démarrage de l'Agent

before_agent s'exécute avant que l'Agent ne commence officiellement son exécution, une seule fois. Vous pouvez y effectuer le prétraitement des entrées, la validation des informations utilisateur, l'initialisation des ressources, etc.

Scénario 1 : Prétraitement des entrées -- Correction automatique de la saisie utilisateur

Exemple

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):
"""Traiter la saisie utilisateur avant le démarrage de l'Agent"""
messages = state.get("messages", [])
if not messages:
return None

# Récupérer le dernier message de l'utilisateur
last_msg = messages[-1]
content = str(last_msg.content) if hasattr(last_msg, 'content') else ""

# Ajouter automatiquement des formules de politesse (si l'utilisateur demande directement)
greetings = ["Hi", "Hello", "hi", "hello", "Hey"]
if content and not any(content.lower().startswith(g) for g in greetings):
# Ne pas modifier, retourner directement
pass

return None

@tool
def search_course(keyword: str) -> str:
"""Rechercher des cours sur Example"""
courses = {
"python": "Tutoriel de base Python3 (gratuit, 30 chapitres)",
"html": "Tutoriel de base HTML (gratuit, 25 chapitres)",
}
return courses.get(keyword.lower(), f"Aucun cours trouvé lié à {keyword}")

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_course],
middleware=[preprocess_input],
system_prompt="Vous êtes le consultant de cours pour Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Cours de Python")]
})
print(f"Réponse : {result['messages'][-1].content}")

Scénario 2 : Contrôle d'accès -- Vérification des permissions

Exemple

from langchain.agents.middleware import before_agent

@before_agent
def access_control(state, runtime):
"""Vérifier si l'utilisateur a l'autorisation d'utiliser l'Agent"""
# Obtenir les informations de l'utilisateur à partir de runtime.context
context = runtime.context
if context is None:
return None

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

# Les utilisateurs invités ne peuvent utiliser que des fonctionnalités limitées
if user_role == "guest":
messages = state.get("messages", [])
if messages:
last_content = str(messages[-1].content)
# Vérifier si cela concerne des fonctionnalités restreintes
restricted_keywords = ["delete", "manage", "configure", "admin"]
if any(kw in last_content for kw in restricted_keywords):
return {
"jump_to": "end",
"messages": [HumanMessage(
content="Vous n'avez pas les autorisations suffisantes pour effectuer cette opération. Veuillez vous connecter et réessayer."
)]
}

return None

after_agent -- Traitement après que l'Agent a terminé

after_agent s'exécute après que l'Agent a terminé tout le traitement (une seule fois). Vous pouvez mettre en forme la sortie finale, enregistrer des informations statistiques, nettoyer les ressources, etc. ici.

Scénario 3: Analyse statistique -- Enregistrement des données de conversation

Exemple

from langchain.agents.middleware import after_agent

@after_agent
def conversation_stats(state, runtime):
"""Compter les informations de conversation et les ajouter au résultat"""
messages = state.get("messages", [])

# Données statistiques
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))

# Envoyer des informations statistiques via un flux personnalisé
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="Vous êtes le consultant de cours pour Example.",
)

# Utilisez stream_mode=["updates", "custom"] pour recevoir des événements personnalisés
for mode, chunk in agent.stream(
{"messages": [HumanMessage(content="Trouver des cours Python")]},
stream_mode=["updates", "custom"],
):
if mode == "custom" and chunk.get("type") == "stats":
print(f"Statistiques : {chunk}")

Sortie :

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

Scénario 4 : Formatage de la sortie -- Unifier le style de réponse

Exemple

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

@after_agent
def format_output(state, runtime):
"""Ajoute des informations récapitulatives formatées au résultat"""
messages = state.get("messages", [])
if not messages:
return None

# Trouver le dernier message IA (réponse finale)
last_ai = None
for msg in reversed(messages):
if msg.type == "ai" and msg.content:
last_ai = msg
break

if last_ai:
# Informations statistiques
tool_msgs = [m for m in messages if m.type == "tool"]
tool_count = len(tool_msgs)

footer = (
f"\n\n---\n"
f"> Cette conversation contenait {len(messages)} messages,"
f"L'outil a été appelé {tool_count} fois.\n"
f"> Propulsé par Example AI Assistant."
)

# Ajouter à la réponse finale
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="Vous êtes le consultant de cours pour Example.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Quels cours propose Python ?")]
})
print(result["messages"][-1].content)

Sortie :

Example offers the Python3 basics tutorial in 30 chapters, completely free and perfect for Python beginners.

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

Exemple complet de collaboration des quatre hooks

Exemple

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

# ----- Définir tous les hooks -----

@before_agent
def init_session(state, runtime):
"""Début : Initialiser la session"""
print(">>> Session démarrée")
return None

@before_model
def pre_model_check(state, runtime):
"""Avant chaque appel au modèle"""
msg_count = len(state.get("messages", []))
print(f" [Avant le modèle] Nombre de messages : {msg_count}")
return None

@after_model
def post_model_check(state, runtime):
"""Après chaque appel au modèle"""
last = state["messages"][-1] if state.get("messages") else None
if last and hasattr(last, 'tool_calls') and last.tool_calls:
print(f" [Après le modèle] Appel d'outil requis")
return None

@after_agent
def finish_session(state, runtime):
"""Fin : nettoyage des ressources"""
total = len(state.get("messages", []))
print(f"<<< Session terminée, {total} messages au total")
return None

# ----- Créer un agent -----

@tool
def get_weather(city: str) -> str:
"""Interroger la météo"""
return f"{city}: Ensoleillé"

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="Vous êtes un assistant.",
)

result = agent.invoke({
"messages": [HumanMessage(content="Quel temps fait-il à Hangzhou ?")]
})
print(f"\n"Réponse finale : {result['messages'][-1].content}")

Sortie :

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

Final reply: Hangzhou is sunny today, great for going out.

Résumé des hooks de middleware

Hook	Nombre d'exécutions	Quand l'utiliser	Capacités clés
before_agent	Une fois	Vérification des permissions, prétraitement des entrées, initialisation des ressources	Peut utiliser jump_to="end" pour terminer prématurément
before_model	À chaque boucle	Troncature des messages, filtrage du contenu, injection de contexte	Peut contrôler le flux avec jump_to
wrap_model_call	À chaque boucle	Nouvelle tentative, repli, mise en cache, modification du prompt	Contrôle total sur l'exécution du modèle
after_model	À chaque boucle	Examen de la réponse, ajout de contenu, journalisation	Peut remplacer la sortie du modèle
wrap_tool_call	Chaque appel d'outil	Nouvelle tentative d'outil, mise en cache, réécriture des paramètres	Contrôle total sur l'exécution des outils
after_agent	Une fois	Mise en forme de la sortie, analyse statistique, nettoyage	Modification de l'état final

Autres extensions

Agent Configuration de développement

Manuel de référence

LangChain @before_agent et @after_agent

before_agent -- Préparation avant le démarrage de l'Agent

Scénario 1 : Prétraitement des entrées -- Correction automatique de la saisie utilisateur

Exemple

Scénario 2 : Contrôle d'accès -- Vérification des permissions

Exemple

after_agent -- Traitement après que l'Agent a terminé

Scénario 3: Analyse statistique -- Enregistrement des données de conversation

Exemple

Scénario 4 : Formatage de la sortie -- Unifier le style de réponse

Exemple

Exemple complet de collaboration des quatre hooks

Exemple

Résumé des hooks de middleware