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. Dans cette leçon, vous apprendrez à rendre vos tools asynchrones et à exploiter le parallélisme natif du SDK.

Tools asynchrones

Le décorateur @function_tool supporte nativement les fonctions async :

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. C’est essentiel dans un serveur web comme FastAPI.

Parallélisme automatique

Le modèle peut décider d’appeler plusieurs tools en parallèle quand les appels sont indépendants :

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.3",
)

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

Si le modèle décide que les deux appels sont indépendants, le SDK les exécute simultanément. Le temps total est celui du tool le plus lent, pas la somme des deux.

Contrôler le parallélisme

Par défaut, le SDK exécute en parallèle tous les tools demandés par le modèle dans un même tour. Vous pouvez contrôler ce comportement :

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.3",
    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. Utile quand l’ordre d’exécution compte.

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

Voici un pattern courant pour agréger des données de plusieurs sources :

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.3",
)

# 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)

Timeouts et résilience

En production, ajoutez des timeouts à vos tools async pour éviter de bloquer l’agent :

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

Le parallélisme peut réduire drastiquement la latence de votre agent :

Séquentiel : 3 tools x 500ms = 1500ms
Parallèle : 3 tools x 500ms = ~500ms (le plus lent)

Pour un agent avec beaucoup de tools, la différence est majeure en termes d’expérience utilisateur.

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