Agent async multi-outils

La boucle d'agent minimale vue précédemment exécute les tool calls dans un for classique : quand le modèle demande plusieurs outils dans la même réponse, la boucle les appelle un par un, en attendant chaque résultat avant de passer au suivant. Pour du calcul pur, la différence est invisible. Pour des outils qui font de l'IO — appel HTTP, requête DB, lecture disque — le temps d'attente de chacun s'additionne au lieu de se chevaucher. asyncio.gather() change ça : les coroutines démarrent ensemble, et le temps total tend vers le plus lent des outils plutôt que vers leur somme.

Rendre les tools awaitable

Le Tool du précédent article pointait vers une fonction Python normale. Pour paralléliser leur exécution, le handler doit devenir une coroutine — une fonction déclarée async def, qui rend la main pendant qu'elle attend une IO au lieu de bloquer le thread :

import asyncio
import json
from dataclasses import dataclass
from typing import Awaitable, Callable, Any


@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict[str, Any]
    handler: Callable[..., Awaitable[str]]

    def to_api(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
        }


async def get_weather(city: str) -> str:
    await asyncio.sleep(0.4)  # ici un vrai appel httpx.AsyncClient vers une API météo
    return json.dumps({"city": city, "temp_c": 21, "condition": "nuageux"})


async def get_stock_price(symbol: str) -> str:
    await asyncio.sleep(0.4)  # même chose : un appel réseau vers une API de marché
    return json.dumps({"symbol": symbol, "price": 187.42})

asyncio.sleep() simule ici la latence réseau qu'un vrai appel httpx.AsyncClient ou aiohttp paierait de toute façon — le point n'est pas le contenu du tool, c'est qu'il rend la main pendant l'attente. Un tool purement CPU (comme calculate dans l'article précédent) n'a rien à gagner à devenir async ; il peut rester synchrone et être appelé via asyncio.to_thread() si on veut uniformiser l'interface sans bloquer la boucle d'événements sur un calcul un peu long.

La boucle : exécuter les tool calls en parallèle

Le changement se joue à l'endroit précis où la boucle d'origine faisait un for block in response.content séquentiel. À la place, on collecte tous les blocs tool_use du tour, puis on les lance ensemble avec asyncio.gather() :

TOOLS = [
    Tool(name="get_weather", description="Météo actuelle d'une ville",
         input_schema={"type": "object", "properties": {"city": {"type": "string"}},
                       "required": ["city"]},
         handler=get_weather),
    Tool(name="get_stock_price", description="Cours actuel d'une action",
         input_schema={"type": "object", "properties": {"symbol": {"type": "string"}},
                       "required": ["symbol"]},
         handler=get_stock_price),
]
TOOLS_BY_NAME = {t.name: t.handler for t in TOOLS}


async def call_tool(block) -> str:
    handler = TOOLS_BY_NAME[block.name]
    try:
        return await handler(**block.input)
    except Exception as exc:
        return f"erreur: {exc}"


async def run_agent(question: str, client, max_turns: int = 6) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]

    for _ in range(max_turns):
        response = await client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            tools=[t.to_api() for t in TOOLS],
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")

        tool_blocks = [b for b in response.content if b.type == "tool_use"]
        outputs = await asyncio.gather(*(call_tool(b) for b in tool_blocks))

        tool_results = [
            {"type": "tool_result", "tool_use_id": block.id, "content": output}
            for block, output in zip(tool_blocks, outputs)
        ]
        messages.append({"role": "user", "content": tool_results})

    return "budget de tours dépassé sans réponse finale"

Trois outils indépendants qui attendent chacun 400 ms passent d'un total de 1200 ms (l'un après l'autre) à environ 400 ms (tous ensemble) — le temps du plus lent, pas la somme :

Idiome à noter : return_exceptions=True

Le code ci-dessus capture déjà les exceptions à l'intérieur de call_tool, donc asyncio.gather() ne voit jamais d'exception remonter. C'est un choix délibéré, et il vaut la peine de comprendre pourquoi : sans ce filet, dès qu'une des coroutines lève une exception, gather() la propage immédiatement à l'endroit du await — mais les coroutines sœurs ne sont pas annulées pour autant. Elles continuent de tourner en arrière-plan, leur résultat (ou leur propre exception) n'étant simplement plus jamais récupéré. C'est une source classique d'avertissements Task exception was never retrieved et de fuites silencieuses.

Deux façons de s'en prémunir : intercepter l'exception dans chaque coroutine avant qu'elle atteigne gather() (l'approche retenue ici, la plus explicite), ou passer return_exceptions=True à gather() pour que chaque résultat soit soit une valeur, soit l'objet exception lui-même, à trier après coup :

resultats = await asyncio.gather(*coros, return_exceptions=True)
sorties = [
    r if not isinstance(r, Exception) else f"erreur: {r}"
    for r in resultats
]

Les deux évitent qu'un tool cassé fasse tomber tout le tour ; le premier garde la logique de gestion d'erreur au plus près du tool concerné, le second la centralise après coup.

Le piège du client synchrone

Convertir les tools en coroutines ne suffit pas : si client est un anthropic.Anthropic classique, l'appel client.messages.create() reste bloquant et doit être remplacé par anthropic.AsyncAnthropic pour être await-able dans une fonction async def. Dans run_agent ci-dessus, un seul appel modèle a lieu par tour — la parallélisation ne le concerne pas directement — mais si l'agent tourne dans un serveur async (FastAPI, par exemple), un appel synchrone bloque tout l'event loop, pas seulement la requête en cours : toutes les autres conversations en cours de traitement s'arrêtent pendant cet appel.

Garde-fous à mettre en place

Tester une boucle async

Le principe du client factice vu précédemment tient toujours, à un détail près : FakeClient.messages.create() doit lui aussi devenir async def, et le test doit être marqué en conséquence. Avec pytest-asyncio :

import pytest

@pytest.mark.asyncio
async def test_run_agent_execute_deux_tools_en_parallele():
    responses = [
        FakeResponse(
            content=[
                FakeBlock(type="tool_use", id="t1", name="get_weather", input={"city": "Lyon"}),
                FakeBlock(type="tool_use", id="t2", name="get_stock_price", input={"symbol": "MSFT"}),
            ],
            stop_reason="tool_use",
        ),
        FakeResponse(content=[FakeBlock(type="text", text="ok")], stop_reason="end_turn"),
    ]
    client = FakeClient(responses)  # .messages.create() est ici une coroutine

    resultat = await run_agent("météo à Lyon et cours MSFT", client)

    assert resultat == "ok"
    tool_results = client.calls[1]["messages"][-1]["content"]
    assert {r["tool_use_id"] for r in tool_results} == {"t1", "t2"}

Ce test ne vérifie pas que l'exécution est effectivement concurrente — un FakeClient sans vraie latence ne peut pas le distinguer d'un for séquentiel. Il vérifie ce qui compte pour la correction : les deux tools sont bien exécutés et leurs deux résultats bien réinjectés, quel que soit l'ordre de complétion des coroutines.

FAQ

Faut-il rendre tous les tools async, même ceux qui sont purement CPU ?

Non. Un tool sans IO n'a rien à gagner à devenir async def — le gain de asyncio.gather() vient du chevauchement des temps d'attente, pas de la parallélisation du calcul. asyncio.to_thread() permet d'appeler une fonction synchrone depuis une boucle async sans la réécrire, utile pour uniformiser l'interface sans changer sa nature.

Que se passe-t-il si un tool lève une exception dans asyncio.gather() ?

Par défaut, l'exception remonte au premier await gather(...), mais les autres coroutines ne sont pas annulées : elles continuent en arrière-plan sans que leur résultat soit récupéré. Passer return_exceptions=True, ou capturer l'exception dans chaque coroutine avant qu'elle atteigne gather(), évite ce comportement.

Le client Anthropic doit-il aussi être async ?

Oui, dès que run_agent est une fonction async def : anthropic.AsyncAnthropic remplace anthropic.Anthropic, sinon l'appel modèle reste bloquant et fige tout l'event loop pendant sa durée — un problème surtout visible si l'agent tourne dans un serveur qui traite plusieurs conversations en même temps.