Middleware LangChain
Le middleware LangChain est la fonctionnalité la plus puissante de LangChain. Il vous permet d'insérer une logique personnalisée à diverses étapes de l'exécution de l'Agent pour implémenter des tentatives de réessai, la dégradation, la mise en cache, le filtrage de contenu, la journalisation et d'autres fonctions — sans modifier le code propre de l'Agent.
Qu'est-ce qu'un middleware
Le middleware est constitué dehooks (Hook)dans le flux d'exécution de l'Agent. Chaque hook vous permet d'exécuter du code personnalisé à des moments précis :
Exemple
# Compréhension intuitive du middleware :
# Supposons que le flux d'exécution de l'Agent soit comme ceci :
# 1. Entrée utilisateur → 2. Réflexion du modèle → 3. Appel éventuel d'outils → 4. Le modèle réfléchit à nouveau → 5. Résultat en sortie
# Le middleware vous permet d'insérer une logique personnalisée entre ces 5 étapes :
# 1. Entrée utilisateur
# ↓ [hook before_agent : journalisation, vérifications de permissions]
# 2. Réflexion du modèle
# ↓ [hook before_model : prétraitement des messages]
# ↓ [hook wrap_model_call : nouvelle tentative, dégradation, mise en cache]
# ↓ [hook after_model : modération du contenu]
# 3. Exécution des outils
# ↓ [hook wrap_tool_call : nouvelle tentative d'appel d'outil]
# 4. Retour à la réflexion du modèle (boucle jusqu'à achèvement)
# ↓ [hook after_agent : formatage des résultats, analyse statistique]
# 5. Résultat de sortie
# Supposons que le flux d'exécution de l'Agent soit comme ceci :
# 1. Entrée utilisateur → 2. Réflexion du modèle → 3. Appel éventuel d'outils → 4. Le modèle réfléchit à nouveau → 5. Résultat en sortie
# Le middleware vous permet d'insérer une logique personnalisée entre ces 5 étapes :
# 1. Entrée utilisateur
# ↓ [hook before_agent : journalisation, vérifications de permissions]
# 2. Réflexion du modèle
# ↓ [hook before_model : prétraitement des messages]
# ↓ [hook wrap_model_call : nouvelle tentative, dégradation, mise en cache]
# ↓ [hook after_model : modération du contenu]
# 3. Exécution des outils
# ↓ [hook wrap_tool_call : nouvelle tentative d'appel d'outil]
# 4. Retour à la réflexion du modèle (boucle jusqu'à achèvement)
# ↓ [hook after_agent : formatage des résultats, analyse statistique]
# 5. Résultat de sortie
Six points de hook
Le middleware de LangChain fournit 6 hooks, répartis en deux catégories selon le moment d'exécution :
| Hook | Fréquence d'exécution | Position d'exécution | Objectif principal |
|---|---|---|---|
| before_agent | Une fois | Avant le démarrage de l'agent | Initialisation, vérifications des autorisations, prétraitement des entrées |
| before_model | À chaque itération de boucle | Avant l'appel au modèle | Prétraitement des messages, injection de contexte dynamique |
| wrap_model_call | À chaque itération de boucle | Encapsule l'appel au modèle | Réessai, dégradation, mise en cache, réécriture de la requête |
| after_model | À chaque itération de boucle | Après l'appel au modèle | Modération de contenu, filtrage des réponses, journalisation |
| wrap_tool_call | À chaque appel d'outil | Encapsule l'exécution de l'outil | Réessai de l'outil, mise en cache des résultats, réécriture des paramètres |
| after_agent | Une fois | Après la fin de l'Agent | Formatage de la sortie, statistiques, nettoyage des ressources |
Deux façons d'utiliser
Un middleware peut être utilisé de deux manières : par héritage de classe ou par décorateurs.
Méthode 1 : Décorateur (recommandé)
Exemple
from langchain.agents.middleware import before_model, after_model
# Approche par décorateur : simple et intuitive
@before_model
def log_before(state, runtime):
"""Journaliser avant chaque appel au modèle"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Nombre actuel de messages : {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Journaliser après chaque appel au modèle"""
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] Le modèle a demandé un appel d'outil")
return None
# Approche par décorateur : simple et intuitive
@before_model
def log_before(state, runtime):
"""Journaliser avant chaque appel au modèle"""
msg_count = len(state.get("messages", []))
print(f"[before_model] Nombre actuel de messages : {msg_count}")
return None
@after_model
def log_after(state, runtime):
"""Journaliser après chaque appel au modèle"""
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] Le modèle a demandé un appel d'outil")
return None
Méthode 2 : Héritage de classe (adapté à une logique complexe)
Exemple
from langchain.agents.middleware import AgentMiddleware
class LoggingMiddleware(AgentMiddleware):
"""Middleware de journalisation personnalisé"""
@property
def name(self) -> str:
# Nom du middleware personnalisé (par défaut, le nom de la classe)
return "logging"
def before_agent(self, state, runtime):
"""Logique avant le démarrage de l'Agent"""
print("[Journalisation] L'Agent a commencé l'exécution")
return None
def before_model(self, state, runtime):
"""Logique avant l'appel au modèle"""
msg_count = len(state.get("messages", []))
print(f"[Journalisation] Préparation de l'appel au modèle, actuellement {msg_count} messages")
return None
def after_model(self, state, runtime):
"""Logique après l'appel au modèle"""
print("[Journalisation] Appel au modèle terminé")
return None
def after_agent(self, state, runtime):
"""Logique après la fin de l'Agent"""
print("[Journalisation] Exécution de l'Agent terminée")
return None
class LoggingMiddleware(AgentMiddleware):
"""Middleware de journalisation personnalisé"""
@property
def name(self) -> str:
# Nom du middleware personnalisé (par défaut, le nom de la classe)
return "logging"
def before_agent(self, state, runtime):
"""Logique avant le démarrage de l'Agent"""
print("[Journalisation] L'Agent a commencé l'exécution")
return None
def before_model(self, state, runtime):
"""Logique avant l'appel au modèle"""
msg_count = len(state.get("messages", []))
print(f"[Journalisation] Préparation de l'appel au modèle, actuellement {msg_count} messages")
return None
def after_model(self, state, runtime):
"""Logique après l'appel au modèle"""
print("[Journalisation] Appel au modèle terminé")
return None
def after_agent(self, state, runtime):
"""Logique après la fin de l'Agent"""
print("[Journalisation] Exécution de l'Agent terminée")
return None
Exemple complet de cycle de vie
Exemple
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):
"""Avant le démarrage de l'Agent"""
print(">>> [before_agent] L'Agent a démarré <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Avant chaque appel au modèle"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Message {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Après chaque appel au modèle"""
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] Outil demandé : {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Réponse directe : {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Après la fin de l'Agent"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Agent terminé, {total_msgs} messages au total <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Interroger la météo"""
return f"{city} : Ensoleillé, 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="Vous êtes un assistant.",
)
print("\n========== Première question (nécessite un outil) ==========")
result = agent.invoke({
"messages": [HumanMessage(content⟦⟧"Quel temps fait-il à Hangzhou ?"="What's the weather in Hangzhou?")]
})
print(f"\nRéponse finale : {result['messages'][-1].content}")
print("\n========== Deuxième question (aucun outil nécessaire) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nRéponse finale : {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):
"""Avant le démarrage de l'Agent"""
print(">>> [before_agent] L'Agent a démarré <<<")
runtime.stream_writer({"type": "lifecycle", "phase": "start"})
return None
@before_model
def pre_model(state, runtime):
"""Avant chaque appel au modèle"""
msg_count = len(state.get("messages", []))
print(f" -> [before_model] Message {msg_count}")
return None
@after_model
def post_model(state, runtime):
"""Après chaque appel au modèle"""
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] Outil demandé : {tools}")
else:
content = str(last.content)[:50] if last and hasattr(last, 'content') else ""
print(f" <- [after_model] Réponse directe : {content}...")
return None
@after_agent
def end_log(state, runtime):
"""Après la fin de l'Agent"""
total_msgs = len(state.get("messages", []))
print(f"<<< [after_agent] Agent terminé, {total_msgs} messages au total <<<")
return None
@tool
def get_weather(city: str) -> str:
"""Interroger la météo"""
return f"{city} : Ensoleillé, 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="Vous êtes un assistant.",
)
print("\n========== Première question (nécessite un outil) ==========")
result = agent.invoke({
"messages": [HumanMessage(content⟦⟧"Quel temps fait-il à Hangzhou ?"="What's the weather in Hangzhou?")]
})
print(f"\nRéponse finale : {result['messages'][-1].content}")
print("\n========== Deuxième question (aucun outil nécessaire) ==========")
result = agent.invoke({
"messages": [HumanMessage(content="Hello")]
})
print(f"\nRéponse finale : {result['messages'][-1].content}")
Sortie :
========== 第一个问题（需要工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 2 条消息 <- [after_model] 请求工具: ['get_weather'] -> [before_model] 第 3 条消息 <- [after_model] 直接回复: 杭州今天晴，气温25°C。... <<< [after_agent] Agent 结束，共 4 条消息 <<< Final reply: 杭州今天晴，气温25°C. ========== 第二个问题（无需工具） ========== >>> [before_agent] Agent 开始 <<< -> [before_model] 第 1 条消息 <- [after_model] 直接回复: 你好！有什么可以帮你的？... <<< [after_agent] Agent 结束，共 2 条消息 <<< Final reply: Hello! How can I help you?
D'après la sortie, vous pouvez voir:
- before_agent after_agent: exécutés une fois par question
- before_model et after_model: exécutés à chaque appel de modèle (la première question a appelé le modèle deux fois, donc chacun a été exécuté deux fois)
Valeur de retour du middleware
La valeur de retour du middleware détermine s'il faut modifier l'état de l'Agent ou contrôler le flux:
| Valeur de retour | Effet | Exemple |
|---|---|---|
| Aucune | Ne modifie aucun état; continue le flux normal | Journalisation uniquement |
| dict | Met à jour l'état de l'Agent (fusionné dans l'état actuel) | Renvoie {"custom_field": "value"} |
| Dictionnaire contenant jump_to | Sauter vers le nœud spécifié | Retourner {"jump_to": "end"} |
Autres extensionsLe dictionnaire retourné est fusionné via le réducteur de l'état de l'Agent. Pour le champ messages, le réducteur add_messages est utilisé, donc les messages retournés sont ajoutés plutôt qu'écrasés.