Renforcer un agent LangChain en Python avec sept middleware intégrés
Découvrez comment le middleware LangChain 1.0 ajoute des fonctionnalités de résumé, des limites de nombre d’appels, des tentatives de récupération, un mécanisme de fallback vers un modèle différent, une suppression des données personnelles et une validation par un humain à un agent Gemini, sans toucher à sa logique de base.
Faire en sorte qu’un agent LangChain réponde aux questions dans un notebook ne prend que quelques minutes. En revanche, obtenir un agent fiable pour un environnement de production est plus difficile : il ne doit pas entrer en boucle au point d’épuiser votre budget API, ni transmettre le numéro de carte bancaire d’un client au fournisseur du modèle, ni envoyer un e-mail non approuvé par quiconque. LangChain 1.0 répond à ces problèmes opérationnels grâce à un middleware, une couche de hooks autour du cycle de l’agent que vous configurez sous forme de simple liste. Ce guide présente d’abord le modèle conceptuel, puis applique sept middlewares à un agent Python basé sur Gemini, avant de conclure avec un middleware personnalisé, afin que vous puissiez voir précisément ce que chaque hook modifie et quand l’utiliser.
Points de contrôle autour du cycle de l’agent
Pensez à un aéroport. L’objectif est de voler d’une ville à une autre, mais autour de cette action principale se trouvent une série de points de contrôle : l’enregistrement confirme l’identité, les scanners de sécurité inspectent les bagages, la porte d’embarquement vérifie les billets, et le récupération des bagages prend le relais après l’atterrissage. Aucun d’eux ne pilote l’avion, et le pilote ne contrôle pas les bagages. Chaque étape effectue une tâche avant ou après l’action principale.
Le middleware applique la même idée aux agents. Le cœur d’un agent est une boucle : appeler le modèle, lui permettre de choisir des outils, les exécuter et répéter jusqu’à ce que le modèle fournisse une réponse finale. Le middleware insère des points de contrôle autour de cette boucle sans la modifier :
- Le suppression des numéros de carte avant que le texte n’atteigne le LLM correspond au scanner de sécurité.
- La nécessité pour une personne d’approuver un e-mail envoyé correspond à la porte d’embarquement.
- L’arrêt après dix appels au modèle pour limiter les dépenses correspond à un disjoncteur.
Si vous travaillez avec TypeScript, l’article associé sur LangChain guardrails and middleware aborde les mêmes concepts du côté JavaScript ; ce guide reste axé sur Python et se concentre sur les classes intégrées concrètes.
Les hooks que le middleware peut utiliser
La boucle met à disposition des hooks à chaque étape, et un middleware peut s’attacher à l’un ou plusieurs d’entre eux :
before_agentetafter_agents’exécutent une fois, au tout début et à la toute fin d’une invocation.before_modeletafter_modelsont déclenchés chaque fois que la boucle est sur le point d’appeler, ou vient de l’appeler, le modèle.wrap_model_calletwrap_tool_callenrobent l’appel lui-même, ce qui leur permet de le réessayer, de le remplacer ou de l’obtenir depuis un cache.
C’est là tout le modèle mental. Vous ajoutez des intermédiaires en passant une liste à create_agent, comme dans cet exemple qui combine la censure des e-mails, une limite de nombre d’appels et des tentatives de réessai des outils :
agent = create_agent(
model=model,
tools=[my_tool],
middleware=[
PIIMiddleware("email", strategy="redact"),
ModelCallLimitMiddleware(run_limit=5),
ToolRetryMiddleware(max_retries=3),
],
)
L’ordre de la liste est important. Les intermédiaires sont appliqués dans l’ordre, comme les couches d’une oignon enveloppant l’agent, de sorte que l’étape de censure listée en premier traite les données brutes avant toute autre étape.
Mise en place et agent de base
Les intermédiaires nécessitent LangChain 1.0 ou une version ultérieure ; installez-le donc avec le flag -U pour mettre à jour toute version plus ancienne. Les exemples utilisent Gemini via son plan gratuit ; vous pouvez créer une clé dans Google AI Studio.
!pip install -qU langchain langchain-google-genai
Importez os ainsi que la classe du modèle de chat Gemini :
import os
from langchain_google_genai import ChatGoogleGenerativeAI
Ensuite, configurez le modèle. Ce fragment définit la clé directement à des fins de démonstration uniquement ; dans du code réel, exportez GOOGLE_API_KEY dans votre environnement ou chargez-la depuis un gestionnaire de secrets plutôt que de la placer dans le code source. Une température nulle permet d’obtenir des résultats reproductibles :
os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)
Le sujet de chaque expérience est un agent minimal sans middleware. Il nécessite la fonction create_agent ainsi que le décorateur tool :
from langchain.agents import create_agent
from langchain_core.tools import tool
Cet agent dispose d’un outil météorologique fictif qui indique toujours du beau temps, et il est invoqué à l’aide d’un seul message de l’utilisateur :
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)
Chaque section ci-dessous ajoute un autre point de contrôle autour de cet agent.
1. SummarizationMiddleware pour une mémoire limitée
Lors de conversations longues, l’historique des messages continue de s’allonger jusqu’à dépasser la fenêtre de contexte, et chaque token supplémentaire est facturé. SummarizationMiddleware vérifie la taille de cet historique dans la fonction before_model ; lorsqu’il dépasse un seuil, il condense les messages plus anciens en un résumé et ne conserve que les messages les plus récents tels quels.
from langchain.agents.middleware import SummarizationMiddleware
La configuration ci-dessous utilise le même modèle pour écrire des résumés, déclenche l’action à 10 messages et conserve les 4 derniers inchangés. Le test crée un historique fictif concernant six villes (douze messages), puis demande quelle ville est apparue en premier :
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
SummarizationMiddleware(
model=model, # which LLM writes the summary
trigger=("messages", 10), # summarize when history hits 10 messages
keep=("messages", 4), # keep the 4 most recent messages intact
),
],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history)) # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"])) # 6
Treize messages sont envoyés et six restent ensuite : le résumé, les quatre messages conservés et la nouvelle réponse. Le modèle répond toujours par « Delhi », car ce fait a été inclus dans le résumé. Compter les messages permet de voir facilement ce comportement lors d’une démonstration, mais en production, un déclencheur basé sur des tokens comme (“tokens”, 3000) ou une fraction de la fenêtre de contexte telle que (“fraction”, 0.8) permet de suivre plus précisément les coûts et d’établir des limites bien meilleures. Gardez à l’esprit que les résumés entraînent des pertes : des chiffres exacts ou des identifiants mentionnés tôt dans la conversation peuvent ne pas être conservés.
2. Les limites de appel en tant que dispositifs de protection des coûts
L’échec le plus coûteux pour un agent est la boucle infinie, où le modèle et les outils continuent de s’appeler mutuellement, ce qui entraîne des dépenses importantes pendant des minutes avant que quiconque ne s’en aperçoive. Deux middlewares imposent une limite stricte à ce phénomène :
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
Ici, le modèle est limité à trois appels par exécution, avec exit_behavior="end" afin que l’agent s’arrête proprement au lieu de lever une exception, et les outils sont limités à deux appels. La demande précise demande délibérément six villes, une par une :
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
# "end" = stop gracefully instead of raising an error
ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
ToolCallLimitMiddleware(run_limit=2),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)
L’agent souhaite effectuer six recherches, atteint ses limites et termine de manière ordonnée avec les résultats partiels qu’il a obtenus. Un paramètre thread_limit est également disponible pour limiter les appels dans l’ensemble d’un thread de conversation plutôt que pour une seule exécution. Ces deux mesures constituent une assurance rudimentaire ; choisissez des limites suffisamment élevées par rapport aux besoins des demandes légitimes afin qu’elles ne s’activent que en cas de dérapages réels.
3. ToolRetryMiddleware pour des dépendances instables
Les outils réels échouent : les appels HTTP expirent et les connexions se coupent. ToolRetryMiddleware utilise wrap_tool_call pour détecter ces échecs et les réessayer avec un retard exponentiel.
from langchain.agents.middleware import ToolRetryMiddleware
Pour le démontrer, un outil de suivi des prix boursiers compte ses tentatives et génère une erreur ConnectionError lors des deux premières avant de réussir. Le middleware permet jusqu’à trois tentatives, avec un délai initial d’une seconde qui double à chaque tentative :
attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
"""Get the current stock price for a ticker symbol."""
attempt_counter["count"] += 1
print(f" [tool called — attempt #{attempt_counter['count']}]")
if attempt_counter["count"] < 3:
raise ConnectionError("API timeout — please retry")
return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
model=model,
tools=[flaky_stock_price],
middleware=[
ToolRetryMiddleware(
max_retries=3, # retry a failed tool up to 3 times
initial_delay=1.0, # wait 1s before first retry
backoff_factor=2.0, # double the wait each time: 1s, 2s, 4s
),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)
L’outil échoue deux fois, le middleware attend puis tente à nouveau, et la troisième tentative réussit. Du point de vue de l’agent, rien ne s’est passé de mal. Les tentatives de réessai ne sont sûres que pour des opérations idempotentes comme les lectures ; réessayer un outil qui charge une carte ou envoie un message pourrait reproduire les effets secondaires.
4. ModelFallbackMiddleware pour les pannes du fournisseur
La même idée de résilience peut protéger les appels au modèle. Lorsque le modèle principal échoue, en raison par exemple d’une limitation de débit ou d’une panne, ce middleware réessaie la requête auprès des modèles de secours dans l’ordre indiqué :
from langchain.agents.middleware import ModelFallbackMiddleware
Dans cet exemple, le modèle Gemini plus léger est utilisé comme principal et un second modèle Gemini est ajouté en tant que secours :
backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
model=model, # primary: gemini-3.5-flash-lite
tools=[get_weather],
middleware=[ModelFallbackMiddleware(backup_model)],
)
Tant que le modèle principal fonctionne correctement, vous ne remarquerez rien, ce qui correspond à l’objectif visé. La valeur n’apparaît que le jour où votre fournisseur rencontre des problèmes. Pour une protection plus forte, envisagez un modèle de secours provenant d’un autre fournisseur, car une panne affecte souvent tous les modèles derrière la même API.
5. PIIMiddleware pour les données sensibles
Fréquemment, vous ne souhaitez pas du tout que des e-mails, des numéros de carte ou des adresses IP soient envoyés au fournisseur du modèle. PIIMiddleware analyse le texte dans la fonction before_model, avant que le modèle ne le voie, et applique l’une de quatre stratégies : redact, mask, hash ou block.
from langchain.agents.middleware import PIIMiddleware
Cet agent ne dispose d’aucun outil. Il remplace complètement les adresses e-mail par un symbole de remplacement et masque les numéros de carte de manière à ne conserver que les quatre derniers chiffres ; ces deux règles s’appliquent aux entrées des utilisateurs :
agent = create_agent(
model=model,
tools=[],
middleware=[
# Replace emails entirely with [REDACTED_EMAIL]
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# Mask credit cards — keeps last 4 digits
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Draft a support reply to priya.sharma@example.com confirming her card "
"4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)
Le modèle rédige sa réponse sans jamais recevoir l’adresse réelle ou le numéro de carte complet. Vous pouvez également enregistrer un type PII personnalisé avec votre propre expression régulière, par exemple pour bloquer tout texte ressemblant à une clé API interne :
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
La détection basée sur des modèles permet de repérer les valeurs correctement formatées, mais pas toutes les orthographes originales ; considérez-la donc comme une couche de protection et non comme une garantie de conformité.
6. HumanInTheLoopMiddleware pour les portes d’approbation
Certaines actions sont trop importantes pour être exécutées de manière totalement autonome : l’envoi d’e-mails, la suppression de données ou le paiement. HumanInTheLoopMiddleware arrête l’agent immédiatement avant que l’outil sensible ne s’exécute, attend une décision humaine puis reprend son fonctionnement.
Cela fonctionne différemment des middlewares précédents. L’agent mis en pause n’est pas terminé. Un checkpointer enregistre son état complet, et l’ID de thread sert de clé pour le retrouver et le reprendre ultérieurement. L’examinateur peut approuver l’appel, modifier ses arguments ou le rejeter.
Ces importations intègrent le middleware, un point de contrôle en mémoire provenant de LangGraph, ainsi que le type Command utilisé pour reprendre l’exécution :
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
L’agent ci-dessous dispose d’une fonction send_email ; il la marque comme pouvant être interrompue et stocke l’état en pause dans InMemorySaver. La première invocation, sur le thread demo-1, demande à l’agent d’envoyer un e-mail à un manager :
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
model=model,
tools=[send_email],
middleware=[
# Pause and ask a human whenever the agent wants to call send_email
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
],
checkpointer=InMemorySaver(), # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
{"messages": [{"role": "user", "content":
"Send an email to boss@company.com saying the report is ready."}]},
config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])
L’exécution s’arrête avant que la fonction ne soit exécutée. L’entrée __interrupt__ dans le résultat montre l’appel en attente avec son destinataire, son sujet et son corps, sans qu’aucun message n’ait été envoyé. L’approbation est alors envoyée sous forme de Command sur le même thread, ce qui permet de reprendre l’exécution en pause :
# Step 2: approve and resume
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config, # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)
Au lieu de approve, vous pouvez envoyer reject avec une raison ou edit avec des arguments modifiés. Dans une application réelle, c’est à ce moment que l’on afficherait une page d’approbation. Deux remarques pratiques : InMemorySaver perd son état lorsque le processus redémarre, donc les systèmes en production ont besoin d’un point de contrôle persistant ; de plus, le format du chargement ultérieur a changé entre les versions de LangChain, il faut donc le vérifier dans la référence sur les middleware correspondant à la version que vous utilisez.
7. Créer vos propres middleware avec un décorateur
Lorsque rien de préconstruit ne convient, un middleware personnalisé reste simple, car chaque point d’ancrage dispose d’un décorateur correspondant. Les importations nécessaires sont le décorateur before_model et le type AgentState :
from langchain.agents.middleware import before_model, AgentState
Cet exemple enregistre le nombre de messages sur le point d’être envoyés à chaque appel au modèle et est ajouté à la liste comme n’importe quel middleware prédéfini :
@before_model
def log_before_model(state: AgentState, runtime) -> None:
print(f" [middleware] Calling model with {len(state['messages'])} messages")
# Returning None = observe only.
# Returning a dict would UPDATE the agent's state (e.g., trim messages)
return Noneagent = create_agent(
model=model,
tools=[get_weather],
middleware=[log_before_model], # plugs in like any prebuilt middleware
)
La valeur de retour constitue un choix de conception important. Retourner None signifie que le middleware se contente d’observer. Retourner un dictionnaire met à jour l’état de l’agent, ce qui permet de filtrer les messages, d’injecter du contexte ou d’imposer des contraintes personnalisées. Il existe également des décorateurs pour les autres points d’intervention : @before_agent, @after_model, @wrap_model_call, @wrap_tool_call, ainsi que @dynamic_prompt pour créer des prompts système à l’exécution.
Points clés
- Le middleware sépare les préoccupations opérationnelles de la logique de l’agent : la boucle reste identique tandis que des points de contrôle sont ajoutés en tant que liste ordinaire transmise à
create_agent.
TodoListMiddleware, LLMToolSelectorMiddleware et ContextEditingMiddleware, est documenté dans la référence officielle.Lectures complémentaires
- De un chatbot à un seul nœud vers un agent géré par MCP dans LangGraph — Construisez une application LangGraph couche par couche : états et réducteurs, arêtes, boucles d’outils, threads avec sauvegardes intermédiaires, trois modes de streaming, et outils fournis via MCP.
- Mémoire d’agent hybride : fusion de BM25 et recherche vectorielle avec RRF en Python — Découvrez pourquoi la recherche vectorielle pure échoue en tant que mémoire d’agent, comment la fusion de rangs réciproques combine BM25 et les résultats densifs en Python, et dans quels cas les résumés GraphRAG sont utiles.