Client asynchrone et performances

Optimiser les performances avec l’asynchrone

Lorsque vous devez générer ou éditer de nombreuses images, les appels séquentiels deviennent un goulot d’étranglement. Le client asynchrone de l’API xAI vous permet d’exécuter plusieurs requêtes en parallèle, réduisant considérablement le temps de traitement total.

Le client AsyncOpenAI

L’asynchrone ne rend pas une génération plus rapide : une image demandée seule mettra exactement le même temps. Ce qu’il supprime, c’est l’attente inutile entre les requêtes — pendant que la première image se calcule côté serveur, votre programme peut en demander dix autres au lieu de rester bloqué. C’est pour cela que le gain est nul sur une image et spectaculaire sur cinquante.

from openai import AsyncOpenAI
import asyncio
import os

async_client = AsyncOpenAI(
    api_key=os.getenv("XAI_API_KEY"),
    base_url="https://api.x.ai/v1"
)

async def generate_image(prompt):
    """Génère une image de manière asynchrone."""
    response = await async_client.images.generate(
        model="grok-imagine-image",
        prompt=prompt,
        resolution="1k"
    )
    return response.data[0].url

async def main():
    url = await generate_image("Un phare sur une falaise, style aquarelle")
    print(url)

asyncio.run(main())

Requêtes parallèles avec asyncio.gather

asyncio.gather lance toutes les requêtes puis attend que l’ensemble soit terminé. La durée totale devient celle de la plus lente, non la somme de toutes — c’est toute la différence, et elle se voit dès trois ou quatre images.

async def generate_batch(prompts):
    """Génère des images pour plusieurs prompts en parallèle."""
    tasks = [generate_image(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks)
    return results

async def main():
    prompts = [
        "Portrait d'un ingénieur, style corporate",
        "Vue panoramique d'une ville futuriste",
        "Diagramme isométrique d'un data center",
        "Logo abstrait bleu et argent",
        "Illustration d'une équipe en réunion"
    ]

    urls = await generate_batch(prompts)
    for prompt, url in zip(prompts, urls):
        print(f"{prompt[:40]}... → {url}")

asyncio.run(main())

Avec 5 prompts, au lieu d’attendre 5 requêtes séquentielles (environ 25 secondes), toutes les images sont générées en parallèle (environ 5-8 secondes).

Contrôler le parallélisme

Sans limite, gather lance tout d’un coup — et cinquante requêtes simultanées sur un modèle plafonné à 30 par minute produisent une avalanche de 429 qui vous coûtera plus de temps que l’exécution séquentielle. Le sémaphore n’est donc pas une précaution optionnelle : il est ce qui rend le parallélisme utilisable. Calibrez-le sur le rate limit du modèle, jamais sur la capacité de votre machine.

async def generate_with_limit(prompts, max_concurrent=10):
    """Génère des images avec un nombre maximum de requêtes parallèles."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_generate(prompt):
        async with semaphore:
            return await generate_image(prompt)

    tasks = [limited_generate(prompt) for prompt in prompts]
    return await asyncio.gather(*tasks)

Pour le modèle standard (300 RPM), vous pouvez utiliser max_concurrent=20 sans problème. Pour le modèle pro (30 RPM), limitez à max_concurrent=5.

Pipeline asynchrone complet

Ce pipeline enchaîne les trois opérations qui doivent aller ensemble : générer, télécharger immédiatement — les URL expirent — et isoler les échecs. Sur un lot important, c’est le troisième point qui fait la différence entre un incident et une perte sèche.

import aiohttp
import aiofiles

async def download_image(session, url, filepath):
    """Télécharge une image de manière asynchrone."""
    async with session.get(url) as resp:
        if resp.status == 200:
            async with aiofiles.open(filepath, "wb") as f:
                await f.write(await resp.read())
            return filepath
    return None

async def pipeline(prompts, output_dir="output"):
    """Pipeline complet : génération + téléchargement."""
    os.makedirs(output_dir, exist_ok=True)

    # Étape 1 : générer toutes les images en parallèle
    urls = await generate_with_limit(prompts, max_concurrent=10)

    # Étape 2 : télécharger toutes les images en parallèle
    async with aiohttp.ClientSession() as session:
        download_tasks = []
        for i, url in enumerate(urls):
            if url:
                filepath = os.path.join(output_dir, f"image_{i:03d}.png")
                download_tasks.append(download_image(session, url, filepath))

        saved_files = await asyncio.gather(*download_tasks)

    count = sum(1 for f in saved_files if f)
    print(f"{count}/{len(prompts)} images générées et sauvegardées")

Comparaison synchrone vs asynchrone

Pour 20 images avec le modèle standard :

• Synchrone : ~100 secondes (20 requêtes x ~5s chacune)
• Asynchrone (max_concurrent=10) : ~10-15 secondes
• Gain : environ 7-10x plus rapide

Gestion des erreurs

Le comportement par défaut de gather est brutal : une exception dans une tâche fait remonter l’erreur et vous perdez les résultats des autres, y compris ceux déjà payés. return_exceptions=True change cela — chaque tâche rend soit son résultat, soit son exception, et vous décidez quoi faire des échecs sans sacrifier les réussites.

async def safe_generate(prompt):
    """Génère une image avec gestion d'erreur."""
    try:
        return await generate_image(prompt)
    except Exception as e:
        print(f"Erreur pour '{prompt[:30]}...' : {e}")
        return None

Points clés à retenir

• Le client AsyncOpenAI permet les requêtes parallèles via asyncio
• Utilisez asyncio.gather pour lancer plusieurs générations simultanément
• Limitez la concurrence avec un sémaphore pour respecter les rate limits
• Le mode asynchrone est 7-10x plus rapide pour les lots importants
• Gérez les erreurs individuellement pour ne pas perdre tout le batch