Interception des appels d'outils LangChain -- @wrap_tool_call
@wrap_tool_call vous permet d'implémenter des capacités de contrôle similaires à @wrap_model_call au niveau de l'exécution des outils — réessai, mise en cache, réécriture des paramètres et post-traitement des résultats.
Structure de base
La structure de @wrap_tool_call est similaire à @wrap_model_call, recevant deux paramètres : request et handler :
Exemple
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call : Contient le nom et les paramètres de l'outil
# request.tool : L'objet outil lui-même
# request.state : L'état actuel de l'Agent
# request.runtime : Contexte d'exécution
# L'outil n'est réellement exécuté que lorsque handler(request) est appelé
result = handler(request)
# result est un ToolMessage ou Command
return result
@wrap_tool_call
def my_tool_wrapper(request, handler):
# request.tool_call : Contient le nom et les paramètres de l'outil
# request.tool : L'objet outil lui-même
# request.state : L'état actuel de l'Agent
# request.runtime : Contexte d'exécution
# L'outil n'est réellement exécuté que lorsque handler(request) est appelé
result = handler(request)
# result est un ToolMessage ou Command
return result
Scénario 1 : Réessai d'appel d'outil
L'exécution d'un outil peut échouer en raison de services externes instables ; les nouvelles tentatives 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_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Réessayer automatiquement lorsqu'un appel d'outil échoue"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Vérifier s'il s'agit d'un résultat d'erreur
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Nouvelle tentative] L'outil a renvoyé une erreur, nouvelle tentative #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Nouvelle tentative réussie] Tentative #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Nouvelle tentative] Exception {e}, nouvelle tentative #{attempt + 1}...")
else:
raise
return last_result
# Simuler un outil susceptible d'échouer
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Obtenir les données du cours depuis EXAMPLE.
Arguments :
course_id : ID du cours
"""
global call_count
call_count += 1
# Simuler un échec pour les deux premières tentatives, succès à la troisième
if call_count < 3:
raise Exception(f"Erreur réseau : impossible de se connecter au service de cours (tentative #{call_count})")
return f"Cours {course_id} : Tutoriel de base Python3, 30 chapitres, gratuit"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Vous êtes l'assistant de cours pour EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Aide-moi à chercher les informations pour le cours python-001")]
})
print(f"\nRéponse finale : {result['messages'][-1].content}")
load_dotenv()
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
@wrap_tool_call
def retry_tool_on_error(request, handler):
"""Réessayer automatiquement lorsqu'un appel d'outil échoue"""
max_retries = 3
last_result = None
for attempt in range(max_retries):
try:
result = handler(request)
# Vérifier s'il s'agit d'un résultat d'erreur
if hasattr(result, 'status') and result.status == "error":
if attempt < max_retries - 1:
print(f" [Nouvelle tentative] L'outil a renvoyé une erreur, nouvelle tentative #{attempt + 1}...")
continue
if attempt > 0:
print(f" [Nouvelle tentative réussie] Tentative #{attempt + 1}")
return result
except Exception as e:
if attempt < max_retries - 1:
import time
time.sleep((attempt + 1) * 2)
print(f" [Nouvelle tentative] Exception {e}, nouvelle tentative #{attempt + 1}...")
else:
raise
return last_result
# Simuler un outil susceptible d'échouer
call_count = 0
@tool
def fetch_course_data(course_id: str) -> str:
"""Obtenir les données du cours depuis EXAMPLE.
Arguments :
course_id : ID du cours
"""
global call_count
call_count += 1
# Simuler un échec pour les deux premières tentatives, succès à la troisième
if call_count < 3:
raise Exception(f"Erreur réseau : impossible de se connecter au service de cours (tentative #{call_count})")
return f"Cours {course_id} : Tutoriel de base Python3, 30 chapitres, gratuit"
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[fetch_course_data],
middleware=[retry_tool_on_error],
system_prompt="Vous êtes l'assistant de cours pour EXAMPLE.",
)
result = agent.invoke({
"messages": [HumanMessage(content="Aide-moi à chercher les informations pour le cours python-001")]
})
print(f"\nRéponse finale : {result['messages'][-1].content}")
Sortie :
[重试] 异常 网络错误：无法连接到课程服务（第 1 次尝试），1 次重试... [重试] 异常 网络错误：无法连接到课程服务（第 2 次尝试），2 次重试... [重试成功] 第 3 次尝试 Final reply: Course python-001 is the Python3 basics tutorial, 30 chapters, free of charge.
Scénario 2 : Modification des paramètres d'outil
La modification dynamique des paramètres avant l'exécution de l'outil permet de transformer les paramètres sans modifier le code de l'outil :
Exemple
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def normalize_city_name(request, handler):
"""Normalise automatiquement les noms de ville (convertir la pleine largeur en demi-largeur, supprimer les espaces supplémentaires, etc.)"""
tool_call = request.tool_call
# Ne traiter que les appels d'outils qui contiennent le paramètre city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Normaliser le nom de ville : supprimer les espaces, unifier la casse
normalized = city.strip().replace(" ", "") # Supprimer les espaces de pleine largeur
# Remplacer le paramètre
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
@wrap_tool_call
def normalize_city_name(request, handler):
"""Normalise automatiquement les noms de ville (convertir la pleine largeur en demi-largeur, supprimer les espaces supplémentaires, etc.)"""
tool_call = request.tool_call
# Ne traiter que les appels d'outils qui contiennent le paramètre city
if "city" in tool_call.get("args", {}):
city = tool_call["args"]["city"]
# Normaliser le nom de ville : supprimer les espaces, unifier la casse
normalized = city.strip().replace(" ", "") # Supprimer les espaces de pleine largeur
# Remplacer le paramètre
new_args = {**tool_call["args"], "city": normalized}
new_tool_call = {**tool_call, "args": new_args}
request = request.override(tool_call=new_tool_call)
return handler(request)
Scénario 3 : Mise en cache des résultats d’outil
Pour les appels d’outil répétés (même outil + mêmes paramètres), les résultats peuvent être mis en cache :
Exemple
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Mettre en cache les résultats d’exécution de l’outil"""
# Générer la clé de cache : nom de l’outil + paramètres
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Vérifier le cache
if cache_key in tool_cache:
print(f"[Succès du cache d'outils] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Exécuter l'outil
result = handler(request)
# Stocker dans le cache
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Écriture dans le cache d'outils] {tool_name}, actuellement {len(tool_cache)} entrées")
return result
from langchain.messages import ToolMessage
tool_cache = {}
@wrap_tool_call
def cache_tool_results(request, handler):
"""Mettre en cache les résultats d’exécution de l’outil"""
# Générer la clé de cache : nom de l’outil + paramètres
tool_name = request.tool_call.get("name", "unknown")
tool_args = str(request.tool_call.get("args", {}))
cache_key = f"{tool_name}:{tool_args}"
# Vérifier le cache
if cache_key in tool_cache:
print(f"[Succès du cache d'outils] {tool_name}")
cached_content = tool_cache[cache_key]
return ToolMessage(
content=cached_content,
tool_call_id=request.tool_call.get("id", ""),
name=tool_name,
)
# Exécuter l'outil
result = handler(request)
# Stocker dans le cache
if hasattr(result, 'content'):
tool_cache[cache_key] = result.content
print(f"[Écriture dans le cache d'outils] {tool_name}, actuellement {len(tool_cache)} entrées")
return result
Scénario 4 : Journalisation et surveillance des appels d'outils
Enregistrer des informations détaillées pour tous les appels d'outils :
Exemple
import time
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Surveiller les métriques de performance des appels d'outils"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Enregistrer l'heure de début
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Enregistrer l'appel réussi
print(f"[Monitor] {tool_name}({tool_args}) a réussi, a pris {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Enregistrer l'appel échoué
print(f"[Monitor] {tool_name}({tool_args}) a échoué, a pris {elapsed:.2f}s, erreur : {e}")
raise
from langchain.agents.middleware import wrap_tool_call
@wrap_tool_call
def monitor_tool_performance(request, handler):
"""Surveiller les métriques de performance des appels d'outils"""
tool_name = request.tool_call.get("name", "unknown")
tool_args = request.tool_call.get("args", {})
# Enregistrer l'heure de début
start_time = time.time()
try:
result = handler(request)
elapsed = time.time() - start_time
# Enregistrer l'appel réussi
print(f"[Monitor] {tool_name}({tool_args}) a réussi, a pris {elapsed:.2f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
# Enregistrer l'appel échoué
print(f"[Monitor] {tool_name}({tool_args}) a échoué, a pris {elapsed:.2f}s, erreur : {e}")
raise
Scénario 5 : Décider du flux suivant en fonction des résultats
Vous pouvez décider s'il faut poursuivre la boucle de l'Agent en fonction du résultat d'exécution de l'outil :
Exemple
from langchain.agents.middleware import wrap_tool_call
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Si l'outil renvoie un résultat vide, terminez directement l'Agent sans gaspiller d'appels au modèle"""
result = handler(request)
# Vérifier si un résultat vide a été renvoyé
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Retourner directement un Command pour mettre à jour l'état
# Ajouter un message AI pour expliquer la situation
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Sorry, no relevant information was found. Please try a different keyword.")
]
})
return result
from langgraph.types import Command
@wrap_tool_call
def check_empty_result(request, handler):
"""Si l'outil renvoie un résultat vide, terminez directement l'Agent sans gaspiller d'appels au modèle"""
result = handler(request)
# Vérifier si un résultat vide a été renvoyé
if hasattr(result, 'content') and (
"Not found" in str(result.content)
or "No results" in str(result.content)
or "None" in str(result.content)
):
# Retourner directement un Command pour mettre à jour l'état
# Ajouter un message AI pour expliquer la situation
from langchain.messages import AIMessage
return Command(update={
"messages": [
AIMessage(content="Sorry, no relevant information was found. Please try a different keyword.")
]
})
return result
Lorsque wrap_tool_call renvoie un Command, vous pouvez modifier l'état de l'Agent via le paramètre update. L'utilisation d'un Command vous permet d'ajouter directement un message AI à la liste des messages, puis la boucle de l'Agent se termine naturellement.
@wrap_model_call vs @wrap_tool_call
| Dimension | @wrap_model_call | @wrap_tool_call |
|---|---|---|
| Cible d'interception | Appel du modèle | Exécution d'outils |
| contenu de la requête | model, messages, tools, system_prompt | tool_call, tool, state, runtime |
| Type de retour | ModelResponse ou AIMessage | ToolMessage ou Command |
| Scénarios applicables | Réessai du modèle, repli, mise en cache, modification du prompt | Réessai d'outil, mise en cache, réécriture des paramètres, traitement des résultats |