Interception des appels de modèle LangChain -- @wrap_model_call
@wrap_model_call est le hook le plus puissant du middleware.
Contrairement à before/after, @wrap_model_call ne se contente pas d'observer ; il peutcontrôler entièrement le processus d'exécution du modèle— réessayer, dégrader, mettre en cache, ou même ignorer le modèle et utiliser directement une réponse prédéfinie.
Comprendre le callback handler
Le cœur de @wrap_model_call est unefonction de callback handlerSeul l'appel à handler(request) exécute réellement le modèle ; si on ne l'appelle pas, le modèle est ignoré.
Exemple
# Structure de base de wrap_model_call
# request : contient toutes les informations telles que le modèle, les messages, les outils, etc.
# handler : un objet appelable ; seule son exécution appelle réellement le modèle
@wrap_model_call
def my_middleware(request, handler):
# Vous pouvez faire n'importe quoi avant l'appel au modèle
print("Le modèle est sur le point d'être appelé...")
# Seul l'appel à handler(request) exécute réellement le modèle
response = handler(request)
# Vous pouvez faire n'importe quoi après l'appel au modèle
print("Appel au modèle terminé")
return response
# request : contient toutes les informations telles que le modèle, les messages, les outils, etc.
# handler : un objet appelable ; seule son exécution appelle réellement le modèle
@wrap_model_call
def my_middleware(request, handler):
# Vous pouvez faire n'importe quoi avant l'appel au modèle
print("Le modèle est sur le point d'être appelé...")
# Seul l'appel à handler(request) exécute réellement le modèle
response = handler(request)
# Vous pouvez faire n'importe quoi après l'appel au modèle
print("Appel au modèle terminé")
return response
Scénario 1 : Mécanisme de réessai
Il s'agit du scénario le plus courant — les appels au modèle peuvent échouer en raison de problèmes réseau, et les réessais automatiques peuvent améliorer la fiabilité :
Exemple
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Réessayer automatiquement lorsque l'appel au modèle échoue, jusqu'à 3 fois"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Réessai réussi] Tentative {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Attente incrémentale : 2s, 4s
print(f" [Réessai] Tentative {attempt + 1} échouée, nouvelle tentative dans {wait_time} secondes...")
import time
time.sleep(wait_time)
# Tous les réessais ont échoué
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Vous êtes l'assistant de EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Présentez le tutoriel Python")]
})
print(f"\nRéponse : {result['messages'][-1].content[:100]}...")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
@wrap_model_call
def retry_on_error(request, handler):
"""Réessayer automatiquement lorsque l'appel au modèle échoue, jusqu'à 3 fois"""
max_retries = 3
last_error = None
for attempt in range(max_retries):
try:
result = handler(request)
if attempt > 0:
print(f" [Réessai réussi] Tentative {attempt + 1}")
return result
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 2 # Attente incrémentale : 2s, 4s
print(f" [Réessai] Tentative {attempt + 1} échouée, nouvelle tentative dans {wait_time} secondes...")
import time
time.sleep(wait_time)
# Tous les réessais ont échoué
raise last_error
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[retry_on_error],
system_prompt="Vous êtes l'assistant de EXAMPLE Tutorial.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Présentez le tutoriel Python")]
})
print(f"\nRéponse : {result['messages'][-1].content[:100]}...")
Scénario 2 : Dégradation du modèle / Basculement
Lorsque le modèle principal est indisponible, basculer automatiquement vers un modèle de secours :
Exemple
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Modèle principal et modèle de secours
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Limiter les tokens pour le modèle dégradé
@wrap_model_call
def fallback_on_error(request, handler):
"""Basculer automatiquement vers le modèle de secours lorsque le modèle principal échoue"""
try:
# Essayer d'utiliser le modèle principal
return handler(request)
except Exception as e:
print(f"[Dégradé] Le modèle principal a échoué : {e}, basculement vers le modèle de secours...")
# Remplacer le modèle dans request, basculer vers le modèle de secours
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Dégradé] Le modèle de secours a également échoué : {e2}")
# Les deux modèles ont échoué, renvoyer un message convivial
return AIMessage(
content="Désolé, le service est temporairement indisponible, veuillez réessayer plus tard."
)
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage
# Modèle principal et modèle de secours
primary_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
fallback_model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0,
max_tokens=100) # Limiter les tokens pour le modèle dégradé
@wrap_model_call
def fallback_on_error(request, handler):
"""Basculer automatiquement vers le modèle de secours lorsque le modèle principal échoue"""
try:
# Essayer d'utiliser le modèle principal
return handler(request)
except Exception as e:
print(f"[Dégradé] Le modèle principal a échoué : {e}, basculement vers le modèle de secours...")
# Remplacer le modèle dans request, basculer vers le modèle de secours
request = request.override(model=fallback_model)
try:
return handler(request)
except Exception as e2:
print(f"[Dégradé] Le modèle de secours a également échoué : {e2}")
# Les deux modèles ont échoué, renvoyer un message convivial
return AIMessage(
content="Désolé, le service est temporairement indisponible, veuillez réessayer plus tard."
)
request.override() est une méthode immuable — elle renvoie une nouvelle copie de request sans modifier l'objet d'origine. Cela garantit que chaque appel est indépendant et sûr.
Scénario 3 : Mise en cache des réponses du modèle
Pour les requêtes répétées, vous pouvez mettre en cache les réponses du modèle afin de réduire les coûts d'appels API :
Exemple
from langchain.agents.middleware import wrap_model_call
from langchain.messages import AIMessage
# Un cache en mémoire simple
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Mettre en cache les réponses du modèle, ne pas appeler de manière répétée pour la même question"""
# Utiliser le contenu du dernier message utilisateur comme clé de cache
messages = request.messages
if not messages:
return handler(request)
# Générer la clé de cache
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Tronquer le contenu trop long
# Vérifier le cache
if cache_key in cache:
print(f"[Succès du cache] Renvoyer directement le résultat mis en cache")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(depuis le cache)*")
# Défaut de cache, appeler le modèle
result = handler(request)
# Le contenu d'AIMessage peut être mélangé dans une liste, prendre directement le premier élément
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Échec de cache] Stocké dans le cache, {len(cache)} entrées actuelles")
elif hasattr(result, 'model_response'):
# Le cas d'ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Vous êtes l'assistant d'EXAMPLE Tutorial et vous devez répondre de manière concise.",
)
# Première requête (échec de cache)
result = agent.invoke({
"messages": [{"role": "user", "content": "Qu'est-ce que Python ?"}]
})
print(f"Première fois : {result['messages'][-1].content[:80]}...\n")
# Deuxième requête avec la même question (succès de cache)
result = agent.invoke({
"messages": [{"role": "user", "content": "Qu'est-ce que Python ?"}]
})
print(f"Deuxième fois : {result['messages'][-1].content[:80]}...")
from langchain.messages import AIMessage
# Un cache en mémoire simple
cache = {}
@wrap_model_call
def cache_responses(request, handler):
"""Mettre en cache les réponses du modèle, ne pas appeler de manière répétée pour la même question"""
# Utiliser le contenu du dernier message utilisateur comme clé de cache
messages = request.messages
if not messages:
return handler(request)
# Générer la clé de cache
last_content = str(messages[-1].content) if hasattr(messages[-1], 'content') else ""
cache_key = last_content[:200] # Tronquer le contenu trop long
# Vérifier le cache
if cache_key in cache:
print(f"[Succès du cache] Renvoyer directement le résultat mis en cache")
cached = cache[cache_key]
return AIMessage(content=f"{cached}\n\n*(depuis le cache)*")
# Défaut de cache, appeler le modèle
result = handler(request)
# Le contenu d'AIMessage peut être mélangé dans une liste, prendre directement le premier élément
if hasattr(result, 'content'):
cache[cache_key] = result.content
print(f"[Échec de cache] Stocké dans le cache, {len(cache)} entrées actuelles")
elif hasattr(result, 'model_response'):
# Le cas d'ExtendedModelResponse
pass
return result
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
middleware=[cache_responses],
system_prompt="Vous êtes l'assistant d'EXAMPLE Tutorial et vous devez répondre de manière concise.",
)
# Première requête (échec de cache)
result = agent.invoke({
"messages": [{"role": "user", "content": "Qu'est-ce que Python ?"}]
})
print(f"Première fois : {result['messages'][-1].content[:80]}...\n")
# Deuxième requête avec la même question (succès de cache)
result = agent.invoke({
"messages": [{"role": "user", "content": "Qu'est-ce que Python ?"}]
})
print(f"Deuxième fois : {result['messages'][-1].content[:80]}...")
Résultat de l'exécution :
[缓存未命中] 已存入缓存，当前 1 条 第一次: Python 是一种高级编程语言，以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言，以简洁易读的语法著称... *(来自缓存)*
Scénario 4 : Modification de la requête — Injection dynamique de messages système
Exemple
from datetime import datetime
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Injecter les informations sur l'heure actuelle avant chaque appel du modèle"""
now = datetime.now()
time_context = (
f"Heure actuelle : {now.strftime('%Y année %m mois %d jour %H:%M')}."
f"Aujourd'hui, c'est {['lundi','mardi','mercredi','jeudi','vendredi','samedi','dimanche'][now.weekday()]}."
)
# Ajouter les informations temporelles au system_message existant
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Utiliser override pour créer une nouvelle requête
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
from langchain.agents.middleware import wrap_model_call
from langchain.messages import SystemMessage
@wrap_model_call
def inject_time_context(request, handler):
"""Injecter les informations sur l'heure actuelle avant chaque appel du modèle"""
now = datetime.now()
time_context = (
f"Heure actuelle : {now.strftime('%Y année %m mois %d jour %H:%M')}."
f"Aujourd'hui, c'est {['lundi','mardi','mercredi','jeudi','vendredi','samedi','dimanche'][now.weekday()]}."
)
# Ajouter les informations temporelles au system_message existant
if request.system_message:
new_content = f"{request.system_message.content}\n\n{time_context}"
else:
new_content = time_context
# Utiliser override pour créer une nouvelle requête
new_request = request.override(
system_message=SystemMessage(content=new_content)
)
return handler(new_request)
Scénario 5 : Combiner plusieurs wrap_model_call
Plusieurs middlewares wrap_model_call sont automatiquement combinés dans l'ordre — le premier se trouve à la couche la plus externe :
Exemple
from langchain.agents.middleware import wrap_model_call
@wrap_model_call
def outer_middleware(request, handler):
"""Middleware le plus externe"""
print("[Externe] Début")
result = handler(request) # Ici, il entre dans inner_middleware
print("[Externe] Fin")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Middleware interne"""
print(" [Interne] Début")
result = handler(request) # C'est ici que le modèle est réellement appelé
print(" [Interne] Fin")
return result
# Ordre d'exécution :
# [Externe] Début
# [Interne] Début
# → Appel réel du modèle
# [Interne] Fin
# [Externe] Fin
@wrap_model_call
def outer_middleware(request, handler):
"""Middleware le plus externe"""
print("[Externe] Début")
result = handler(request) # Ici, il entre dans inner_middleware
print("[Externe] Fin")
return result
@wrap_model_call
def inner_middleware(request, handler):
"""Middleware interne"""
print(" [Interne] Début")
result = handler(request) # C'est ici que le modèle est réellement appelé
print(" [Interne] Fin")
return result
# Ordre d'exécution :
# [Externe] Début
# [Interne] Début
# → Appel réel du modèle
# [Interne] Fin
# [Externe] Fin
Autres extensionsPlusieurs appels à wrap_model_call s'enveloppent couche par couche comme un oignon. La couche la plus externe s'exécute en premier et retourne en dernier. Cela vous permet de combiner plusieurs fonctionnalités indépendantes — par exemple, la couche externe effectue la vérification du cache, la couche interne effectue les nouvelles tentatives, sans interférer entre elles.