Tools asynchrones et parallèles

En production, vos tools appellent des API externes, interrogent des bases de données ou effectuent des opérations I/O. Ces opérations prennent du temps, et ce temps s’additionne : un agent qui enchaîne cinq appels d’une seconde met cinq secondes à répondre. Rendre les tools asynchrones et exploiter le parallélisme natif du SDK est ce qui sépare une démonstration d’un service utilisable.

Tools asynchrones

Le décorateur @function_tool supporte nativement les fonctions async. Rien à configurer : vous déclarez async def, vous utilisez un client HTTP asynchrone, et le SDK s’occupe du reste.

import httpx
from agents import function_tool

@function_tool
async def rechercher_entreprise(nom: str) -> str:
    """Recherche des informations sur une entreprise via l'API Societe.com."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.societe.com/v1/search",
            params={"q": nom},
            timeout=10,
        )
        if response.status_code == 200:
            data = response.json()
            return f"Entreprise : {data['name']}, SIREN : {data['siren']}, CA : {data['revenue']}"
        return f"Aucun résultat pour '{nom}'"

@function_tool
async def verifier_solvabilite(siren: str) -> str:
    """Vérifie la solvabilité d'une entreprise par son numéro SIREN."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.credit-scoring.com/v1/{siren}",
            timeout=10,
        )
        data = response.json()
        return f"Score de solvabilité : {data['score']}/100, risque : {data['risk_level']}"

Quand le Runner exécute un tool async, il ne bloque pas la boucle événementielle. Dans un serveur FastAPI, cela signifie que les autres requêtes continuent d’être traitées pendant que votre agent attend la réponse de l’API Societe.com.

Parallélisme automatique

Deux tools asynchrones et indépendants peuvent partir en même temps. Le modèle en décide, et les instructions l’y invitent explicitement.

from agents import Agent, Runner

agent = Agent(
    name="Analyste B2B",
    instructions="""Vous analysez des entreprises pour des décisions commerciales.
    Quand on vous demande d'analyser une entreprise, récupérez ses informations
    ET vérifiez sa solvabilité en même temps.""",
    tools=[rechercher_entreprise, verifier_solvabilite],
    model="gpt-5.6-terra",
)

# Le modèle lancera les deux tools en parallèle
result = await Runner.run(
    agent,
    "Analysez l'entreprise Acme Corp (SIREN 123456789)"
)

Le SIREN étant fourni dans la question, la vérification de solvabilité n’attend rien de la recherche d’entreprise : le SDK exécute les deux appels simultanément. Le temps total est celui du tool le plus lent, pas la somme des deux.

Contrôler le parallélisme

Le parallélisme n’est pas toujours souhaitable. Si vos tools écrivent dans un même enregistrement ou si l’ordre conditionne la cohérence des données, forcez la séquence.

from agents import Agent

agent = Agent(
    name="Agent prudent",
    instructions="Exécutez les outils un par un pour garantir la cohérence des données.",
    tools=[rechercher_entreprise, verifier_solvabilite],
    model="gpt-5.6-terra",
    parallel_tool_calls=False,  # Force l'exécution séquentielle
)

Le paramètre parallel_tool_calls=False force le modèle à n’appeler qu’un seul tool à la fois. Vous perdez en latence ce que vous gagnez en prévisibilité.

Pattern : agrégation de résultats parallèles

Le cas le plus fréquent en pratique consiste à réunir des données de plusieurs sources pour une même entité. Ici, la météo et l’agenda culturel d’une ville n’ont rien à voir l’un avec l’autre, mais la réponse a besoin des deux.

import json
from agents import Agent, Runner, function_tool

@function_tool
async def obtenir_meteo(ville: str) -> str:
    """Obtient la météo actuelle pour une ville."""
    # Simulation - en production : appel API météo
    meteos = {
        "Paris": {"temp": 22, "conditions": "ensoleillé"},
        "Lyon": {"temp": 19, "conditions": "nuageux"},
        "Marseille": {"temp": 26, "conditions": "ensoleillé"},
    }
    data = meteos.get(ville, {"temp": 0, "conditions": "inconnu"})
    return json.dumps({"ville": ville, **data}, ensure_ascii=False)

@function_tool
async def obtenir_evenements(ville: str) -> str:
    """Obtient les événements à venir dans une ville."""
    evenements = {
        "Paris": ["Concert au Stade de France", "Exposition au Louvre"],
        "Lyon": ["Festival des Lumières", "Salon Gastronomique"],
        "Marseille": ["Régate de la Méditerranée"],
    }
    return json.dumps({
        "ville": ville,
        "evenements": evenements.get(ville, [])
    }, ensure_ascii=False)

agent = Agent(
    name="Agent voyage",
    instructions="""Vous êtes un agent de voyage. Quand on vous demande
    des recommandations pour une ville, récupérez la météo ET les événements
    simultanément pour donner un conseil complet.""",
    tools=[obtenir_meteo, obtenir_evenements],
    model="gpt-5.6-terra",
)

# Le modèle appellera les deux tools en parallèle pour chaque ville
result = await Runner.run(
    agent,
    "Je veux visiter Paris ce week-end. Que me conseillez-vous ?"
)
print(result.final_output)

Les deux tools retournent du JSON, ce qui facilite la recombinaison par le modèle. Notez la formulation des instructions : le « ET … simultanément » est ce qui déclenche l’appel parallèle plutôt qu’un enchaînement paresseux.

Timeouts et résilience

Une API qui ne répond jamais bloque votre agent aussi sûrement qu’un tool synchrone. Bornez chaque appel lent avec un timeout et transformez l’échec en message intelligible.

import asyncio
from agents import function_tool

@function_tool
async def appel_api_lent(query: str) -> str:
    """Appelle une API qui peut être lente."""
    try:
        async with asyncio.timeout(5):  # Timeout de 5 secondes
            # Votre appel API ici
            await asyncio.sleep(0.1)  # Simulation
            return f"Résultat pour '{query}'"
    except asyncio.TimeoutError:
        return "L'API n'a pas répondu dans les temps. Veuillez réessayer."

Performance : impact du parallélisme

L’ordre de grandeur mérite d’être posé. Trois tools de 500 ms exécutés séquentiellement coûtent 1500 ms ; les mêmes trois tools en parallèle coûtent environ 500 ms, soit la durée du plus lent. Pour un agent qui mobilise beaucoup d’outils, la différence se voit immédiatement à l’usage.

Points clés à retenir

Utilisez async pour les tools qui font des appels réseau ou I/O
Le SDK exécute automatiquement en parallèle les tools indépendants
parallel_tool_calls=False force l’exécution séquentielle si nécessaire
Ajoutez des timeouts avec asyncio.timeout() pour la résilience
Le parallélisme peut diviser la latence par le nombre de tools