14. Construire la boucle de l'agent
Au chapitre 13, nous avons appris à exécuter les demandes d'appel d'outil du LLM et à renvoyer les résultats. Cependant, ce travail supposait que chaque tâche pouvait être traitée en un seul appel d'outil. En pratique, les appels d'outils ont souvent besoin de se poursuivre sur plusieurs tours jusqu'à ce que le LLM ait rassemblé toutes les informations nécessaires pour une réponse finale.
Prenons la demande « trouve la population de la France, puis multiplie-la par deux ». Cette tâche nécessite au moins deux appels d'outils. Vous devez d'abord rechercher la population — ce n'est qu'ensuite que vous pouvez effectuer le calcul. Le deuxième appel dépend du premier résultat, il n'y a donc aucun moyen de le traiter en un seul appel d'outil.
Dans ce chapitre, nous transformons le cycle unique du chapitre 13 en une boucle. Tant que le LLM demande des appels d'outils, nous continuons à les exécuter — en répétant jusqu'à ce que le LLM cesse de lui-même de demander des outils. Puis nous ajoutons des limites de sécurité pour empêcher la boucle de tourner indéfiniment, et abordons la gestion des erreurs afin que l'agent ne plante pas lorsqu'un outil échoue.
14.1) Du cycle unique à la boucle
14.1.1) Comment fonctionne la boucle de l'agent ?
Comme nous l'avons vu dans l'introduction, les tâches où l'action suivante dépend du résultat de l'étape précédente ne peuvent souvent pas être traitées avec un seul appel d'outil. Le LLM doit appeler un outil, vérifier le résultat et décider à nouveau. C'est ce que fait la boucle de l'agent, et elle fonctionne en trois étapes :
- Réfléchir (Think) — Le LLM lit la conversation jusqu'à présent et décide de la prochaine action. S'il a besoin d'un outil, il demande un appel d'outil via
tool_calls. Sinon, il renvoie une réponse finale. - Agir (Act) — Exécute l'outil spécifié dans
tool_calls. - Observer (Observe) — Vérifie le résultat de l'exécution de l'outil et l'ajoute à la conversation dans un
ToolMessage.
La boucle de l'agent répète ces trois étapes jusqu'à ce que le LLM ne demande plus aucun outil. Ce modèle est également connu sous le nom de ReAct (Reason + Act), et l'idée clé est d'alterner entre le raisonnement et l'action.
Si tool_calls est présent, exécutez l'outil, ajoutez le résultat à la conversation et appelez à nouveau le LLM. Si tool_calls est vide, le LLM a renvoyé une réponse finale et la boucle se termine. Passons maintenant au code.
14.1.2) Implémenter la boucle Réfléchir-Agir-Observer
Transformons la boucle Réfléchir-Agir-Observer de la section précédente en code. Tout d'abord, nous préparons les outils et le modèle.
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain.tools import tool
# Définit les outils
@tool
def get_weather(city: str) -> str:
"""Obtient la météo actuelle pour une ville."""
fake_data = {"Tokyo": "18°C, nuageux", "Le Caire": "31°C, ensoleillé"}
return fake_data.get(city, f"Aucune donnée météo pour {city}.")
@tool
def calculate(expression: str) -> str:
"""Évalue une expression arithmétique simple, par ex. '3 * 21'."""
return str(eval(expression)) # Attention : eval() est un risque de sécurité. Ne pas utiliser en production.
# Lie les outils
tools = [get_weather, calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}Au chapitre 13, nous exécutions un outil une seule fois puis nous arrêtions. Maintenant, nous répétons jusqu'à ce que le LLM cesse de demander des appels d'outils. À l'intérieur d'une boucle while True, nous appelons le LLM, et si la réponse contient des tool_calls, nous exécutons les outils et appelons à nouveau le LLM. S'il n'y a pas de tool_calls, le LLM a renvoyé une réponse finale, nous sortons donc de la boucle.
def run_agent(user_input: str) -> str:
"""Exécute la boucle Réfléchir-Agir-Observer jusqu'à ce que le LLM renvoie une réponse finale."""
messages = [
SystemMessage(content="Vous êtes un assistant utile."),
HumanMessage(content=user_input),
]
while True:
# RÉFLÉCHIR : Demande au LLM de décider de la prochaine action
print("RÉFLÉCHIR : Demande au LLM de décider")
ai_message = llm_with_tools.invoke(messages)
messages.append(ai_message)
# Condition de sortie : pas de tool_calls signifie que c'est la réponse finale
if not ai_message.tool_calls:
return ai_message.content
# AGIR + OBSERVER : Exécute les outils demandés et ajoute les résultats à la conversation
for tool_call in ai_message.tool_calls:
selected_tool = tool_map[tool_call["name"]]
print(f"AGIR : appel de '{tool_call['name']}', args={tool_call['args']}")
tool_message = selected_tool.invoke(tool_call)
print(f"OBSERVER : {tool_message.content}")
messages.append(tool_message)Comparons cela avec le code du chapitre 13. Au chapitre 13, après avoir exécuté les outils, nous appelions le LLM une dernière fois pour obtenir la réponse. Au chapitre 14, nous plaçons ce même processus à l'intérieur d'un while True et vérifions tool_calls à chaque itération pour décider s'il faut continuer. Les briques de base sont les mêmes qu'au chapitre 13 — nous les avons simplement enveloppées dans une boucle.
Exécutons-le.
answer = run_agent("Quel temps fait-il à Tokyo, et fait-il assez chaud pour se promener ?")
print(f'Réponse finale : {answer}')Sortie :
RÉFLÉCHIR : Demande au LLM de décider
AGIR : appel de 'get_weather', args={'city': 'Tokyo'}
OBSERVER : 18°C, nuageux
RÉFLÉCHIR : Demande au LLM de décider
Réponse finale : En ce moment à Tokyo, il fait 18°C (environ 64°F) et le temps est nuageux.
Cette température est généralement douce et agréable pour se promener pour la plupart des gens.La sortie montre le flux RÉFLÉCHIR → AGIR → OBSERVER → RÉFLÉCHIR. Lors de la première itération, le LLM a demandé un appel get_weather, et lors de la deuxième itération, il a vu le résultat météo et a généré une réponse finale. Comme tool_calls était vide, la réponse finale a été renvoyée et la boucle s'est terminée.
Testons maintenant le scénario d'étapes dépendantes de l'introduction — où l'appel suivant ne peut avoir lieu qu'après avoir vu le résultat du précédent.
answer = run_agent("Obtiens la température au Caire, puis multiplie le nombre par 3.")
print(f'Réponse finale : {answer}')Sortie :
RÉFLÉCHIR : Demande au LLM de décider
AGIR : appel de 'get_weather', args={'city': 'Le Caire'}
OBSERVER : 31°C, ensoleillé
RÉFLÉCHIR : Demande au LLM de décider
AGIR : appel de 'calculate', args={'expression': '31 * 3'}
OBSERVER : 93
RÉFLÉCHIR : Demande au LLM de décider
Réponse finale : Température actuelle au Caire : 31°C. Multipliée par 3 = 93.Cette fois, la boucle a effectué trois itérations.
- Première itération — Le LLM demande
get_weather("Le Caire"). - Deuxième itération — Après avoir vu le résultat
"31°C, ensoleillé", le LLM demandecalculate("31 * 3"). Il n'a pu former l'expression qu'après avoir vu la température. - Troisième itération — Après avoir vu les deux résultats
"31°C, ensoleillé"et"93", le LLM a renvoyé la réponse finale.
14.2) Ajouter des limites de sécurité
La boucle que nous avons construite ci-dessus n'a qu'une seule condition de sortie : lorsque le LLM répond sans tool_calls, nous sortons de la boucle. Dans des circonstances normales, cela suffit, mais que se passe-t-il si le LLM ne cesse jamais de demander des appels d'outils ?
Par exemple, si un outil renvoie toujours des résultats ambigus, le LLM peut continuer à l'appeler en espérant obtenir quelque chose de mieux. Comme la boucle est while True, si le LLM ne s'arrête pas, le programme ne s'arrête pas non plus. Les coûts d'appels d'API continuent de s'accumuler tandis que le programme s'exécute indéfiniment.
La solution la plus simple est de fixer un plafond au nombre de fois où la boucle peut s'exécuter. Remplacez while True par for step in range(max_steps), et la boucle est garantie de se terminer après max_steps itérations, quel que soit le comportement du LLM.
def run_agent(user_input: str, max_steps: int = 10) -> str:
"""Exécute la boucle Réfléchir-Agir-Observer dans la limite de max_steps itérations."""
messages = [
SystemMessage(content="Vous êtes un assistant utile."),
HumanMessage(content=user_input),
]
for step in range(max_steps):
# RÉFLÉCHIR
ai_message = llm_with_tools.invoke(messages)
messages.append(ai_message)
# Condition de sortie : pas de tool_calls signifie que c'est la réponse finale
if not ai_message.tool_calls:
return ai_message.content
# AGIR + OBSERVER
for tool_call in ai_message.tool_calls:
selected_tool = tool_map[tool_call["name"]]
tool_message = selected_tool.invoke(tool_call)
messages.append(tool_message)
# max_steps atteint : la boucle s'est terminée sans réponse finale
return f"[Arrêté après avoir atteint le nombre maximal d'itérations ({max_steps})]"Par rapport au code précédent, deux choses ont changé. while True est devenu for step in range(max_steps), et une valeur de retour a été ajoutée pour le cas où la boucle atteint max_steps. La boucle se termine désormais de l'une des deux façons suivantes : le LLM renvoie une réponse finale de lui-même (terminaison naturelle), ou max_steps est atteint (terminaison de sécurité).
Vérifions que la limite de sécurité fonctionne réellement. Nous allons créer un outil qui ne renvoie jamais de résultats utiles, forçant le LLM dans une situation où il ne cesse jamais de demander des appels d'outils.
@tool
def unhelpful_search(query: str) -> str:
"""Recherche des informations."""
return "Aucun résultat trouvé. Essayez de reformuler votre requête."
llm_with_bad_tool = llm.bind_tools([unhelpful_search])
tool_map_bad = {unhelpful_search.name: unhelpful_search}
def run_agent_bad(user_input: str, max_steps: int = 5) -> str:
messages = [
SystemMessage(content=(
"Vous devez TOUJOURS utiliser l'outil unhelpful_search pour trouver des informations. "
"Vous n'êtes PAS autorisé à répondre à partir de vos propres connaissances. "
"Si l'outil ne renvoie aucun résultat, vous DEVEZ reformuler et rechercher à nouveau. "
"Continuez à chercher jusqu'à ce que vous trouviez la réponse."
)),
HumanMessage(content=user_input),
]
for step in range(max_steps):
print(f"--- Étape {step + 1} ---")
ai_message = llm_with_bad_tool.invoke(messages)
messages.append(ai_message)
if not ai_message.tool_calls:
return ai_message.content
for tool_call in ai_message.tool_calls:
selected_tool = tool_map_bad[tool_call["name"]]
tool_message = selected_tool.invoke(tool_call)
print(f"AGIR : '{tool_call['name']}' → {tool_message.content}")
messages.append(tool_message)
return f"[Arrêté après avoir atteint le nombre maximal d'itérations ({max_steps})]"
answer = run_agent_bad("Quelle est la population de la France ?")
print(f"\nRéponse finale : {answer}")Sortie :
--- Étape 1 ---
AGIR : 'unhelpful_search' → Aucun résultat trouvé. Essayez de reformuler votre requête.
--- Étape 2 ---
AGIR : 'unhelpful_search' → Aucun résultat trouvé. Essayez de reformuler votre requête.
--- Étape 3 ---
AGIR : 'unhelpful_search' → Aucun résultat trouvé. Essayez de reformuler votre requête.
--- Étape 4 ---
AGIR : 'unhelpful_search' → Aucun résultat trouvé. Essayez de reformuler votre requête.
--- Étape 5 ---
AGIR : 'unhelpful_search' → Aucun résultat trouvé. Essayez de reformuler votre requête.
Réponse finale : [Arrêté après avoir atteint le nombre maximal d'itérations (5)]Sans max_steps, cette boucle aurait tourné indéfiniment. Grâce à max_steps=5, elle a été forcée de s'arrêter après cinq itérations.
La bonne valeur pour max_steps dépend de la complexité de votre agent. Trop basse, et les tâches complexes sont interrompues prématurément. Trop haute, et un agent qui se comporte mal accumule des coûts avant d'être arrêté. 15–25 est un point de départ courant ; ajustez en fonction de vos charges de travail réelles.
14.3) Gérer les erreurs d'outils dans la boucle
La boucle que nous avons construite en 14.1 et 14.2 suppose que les outils s'exécutent toujours avec succès. Mais que se passe-t-il si un outil lève une exception ? Le code actuel n'a aucune gestion d'exception, donc si une exception survient pendant l'exécution de l'outil, l'agent entier plante.
Au chapitre 12, nous avons appris à intercepter les exceptions à l'intérieur de l'outil lui-même avec try/except et à renvoyer les messages d'erreur sous forme de chaînes. Si un outil est construit de cette façon, il n'y a pas de problème. Mais tous les outils ne gèrent pas les erreurs en interne. Les outils qui appellent des bibliothèques ou des API externes peuvent lever des exceptions inattendues.
Pour se prémunir contre cela, il est judicieux de gérer les erreurs au niveau de la boucle également. L'approche est simple : enveloppez l'exécution de l'outil dans un try/except, et si une exception survient, placez le message d'erreur dans un ToolMessage et transmettez-le au LLM. Le LLM peut lire ce message d'erreur et réessayer avec des arguments corrigés ou choisir une approche différente. C'est ce qu'on appelle l'auto-correction (self-correction).
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
# Définit les outils
@tool
def calculate(expression: str) -> str:
"""Évalue une expression arithmétique simple, par ex. '3 * 21'."""
return str(eval(expression)) # Attention : eval() est un risque de sécurité. Ne pas utiliser en production.
# Lie les outils
tools = [calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}
def run_agent(user_input: str, max_steps: int = 10) -> str:
"""Boucle de l'agent qui transmet les erreurs d'outils au LLM, permettant l'auto-correction."""
messages = [
SystemMessage(content="Vous êtes un assistant utile."),
HumanMessage(content=user_input),
]
for step in range(max_steps):
# RÉFLÉCHIR
print("RÉFLÉCHIR : Demande au LLM de décider")
ai_message = llm_with_tools.invoke(messages)
messages.append(ai_message)
if not ai_message.tool_calls:
return ai_message.content
# AGIR + OBSERVER
for tool_call in ai_message.tool_calls:
selected_tool = tool_map[tool_call["name"]]
try:
print(f"AGIR : appel de '{tool_call['name']}', args={tool_call['args']}")
tool_message = selected_tool.invoke(tool_call)
print(f"OBSERVER : {tool_message.content}")
except Exception as e:
print(f"OBSERVER : Erreur - {e}")
tool_message = ToolMessage(
content=f"Erreur : {e}",
tool_call_id=tool_call["id"],
)
messages.append(tool_message)
return f"[Arrêté après avoir atteint le nombre maximal d'itérations ({max_steps})]"Par rapport au code de 14.2, le seul changement est le try/except. Si un outil lève une exception, le message d'erreur est placé dans un ToolMessage et ajouté à la conversation. Notez que même un appel échoué doit avoir un ToolMessage avec le tool_call_id correspondant. Le LLM verra cette erreur lors de l'itération suivante et décidera de la marche à suivre.
Vérifions que l'auto-correction fonctionne. Nous allons déclencher une exception en demandant à l'outil calculate de diviser par zéro.
answer = run_agent("Utilise l'outil calculatrice pour calculer 10 / 0")
print(f"Réponse finale : {answer}")Sortie :
RÉFLÉCHIR : Demande au LLM de décider
AGIR : appel de 'calculate', args={'expression': '10 / 0'}
OBSERVER : Erreur - division by zero
RÉFLÉCHIR : Demande au LLM de décider
Réponse finale : J'ai utilisé l'outil calculatrice et il a renvoyé une erreur : « division by zero ».
Explication : 10 / 0 n'est pas défini en arithmétique ordinaire, il ne peut donc pas produire un nombre fini.
Souhaitez-vous que je :
- calcule les limites à gauche et à droite,
- affiche le résultat en virgule flottante IEEE-754,
- ou évalue une expression différente ?Lors de la première itération, le LLM a demandé calculate("10 / 0") et une ZeroDivisionError a été levée. Le try/except a intercepté l'exception et transmis le message d'erreur au LLM. Lors de la deuxième itération, le LLM a vu le message d'erreur et a renvoyé une réponse finale expliquant que la division par zéro est impossible. Sans le try/except, le programme aurait planté à la première ZeroDivisionError.