Tu as un script qui interroge une API, attend la réponse, puis interroge la suivante. Avec un LLM qui met deux secondes à générer sa réponse et cinquante prompts à traiter, tu passes cent secondes à fixer ton terminal. asyncio ne rend pas chaque réponse plus rapide : il arrête de les attendre l'une après l'autre. Pendant que la première requête dort, les quarante-neuf autres sont déjà parties.
asyncio, c'est la bibliothèque standard de Python pour la concurrence. Pas de threads, pas de processus : une boucle d'événements (event loop) qui jongle entre des tâches. Dès qu'une tâche doit patienter sur le réseau, elle se met en pause et laisse la main aux autres. Une seule idée à retenir : await veut dire « je rends la main, réveille-moi quand c'est prêt ».
Étape 1 : la fonction qui se met en pause
Une coroutine se déclare avec async def. À l'intérieur, await marque le point où la fonction accepte d'attendre sans bloquer tout le reste du programme.
import asyncio
async def simule_requete(nom, duree):
await asyncio.sleep(duree) # attendre SANS bloquer
return f"{nom} terminé"asyncio.sleep joue ici le rôle d'un vrai appel réseau : pendant ces deux secondes, Python est libre de faire autre chose. La coroutine ne s'exécute pas toute seule, on la démarre depuis la racine avec asyncio.run.
Étape 2 : le piège de l'await en boucle
Si tu await chaque tâche dans une boucle, tu retombes exactement dans le séquentiel : dix appels d'une seconde, dix secondes au compteur.
import asyncio
import time
async def simule_requete(nom, duree):
await asyncio.sleep(duree)
return f"{nom} terminé"
async def en_serie():
for i in range(10):
await simule_requete(f"tâche {i}", 1)
start = time.perf_counter()
asyncio.run(en_serie())
print(f"Série : {time.perf_counter() - start:.2f} s") # ~10 sDix secondes. L'await suspend la boucle à chaque itération, comme un appel classique. Autant ne pas se fatiguer.
Étape 3 : asyncio.gather pour tout lancer d'un coup
Le bon réflexe : créer toutes les coroutines d'abord (sans les attendre), puis les confier à asyncio.gather. gather les démarre toutes en même temps et ne rend la main qu'une fois la dernière terminée.
async def en_parallele():
taches = [simule_requete(f"tâche {i}", 1) for i in range(10)]
return await asyncio.gather(*taches)
start = time.perf_counter()
asyncio.run(en_parallele())
print(f"Parallèle : {time.perf_counter() - start:.2f} s") # ~1 sUne seconde au lieu de dix. Sur ma machine, le script affiche Série : 10.01 s puis Parallèle : 1.00 s. Le gain est exactement le nombre de tâches, tant que le goulot d'étranglement vient du temps d'attente et pas de ton CPU.
Étape 4 : de vraies requêtes HTTP en parallèle avec aiohttp
Pour interroger une vraie API de façon asynchrone, il te faut un client HTTP asynchrone : aiohttp fait le travail. On récupère quatre réponses lentes, une seconde chacune, en un seul aller-retour.
import asyncio
import time
import aiohttp
async def recupere(session, url):
async with session.get(url, timeout=10) as rep:
return url, rep.status
async def main():
urls = ["https://httpbin.org/delay/1"] * 4
async with aiohttp.ClientSession() as session:
taches = [recupere(session, u) for u in urls]
resultats = await asyncio.gather(*taches)
for url, statut in resultats:
print(statut, url)
start = time.perf_counter()
asyncio.run(main())
print(f"4 requêtes lentes : {time.perf_counter() - start:.2f} s")Quatre requêtes qui dorment une seconde chacune, expédiées en un seul round. Remplace httpbin par ton endpoint : le principe est le même pour cinquante prompts envoyés à ton LLM en parallèle.
Étape 5 : asyncio.to_thread pour ton vieux code synchrone
Tout ton code n'est pas prêt pour async. Bonne nouvelle : pas besoin de tout réécrire. asyncio.to_thread expédie une fonction bloquante dans un thread séparé et te rend la main aussitôt. Parfait pour requests, un SDK uniquement synchrone ou n'importe quelle bibliothèque qui bloque.
import asyncio
import requests
def requete_bloquante(url):
return url, requests.get(url, timeout=10).status_code
async def main():
urls = ["https://httpbin.org/delay/1"] * 4
resultats = await asyncio.gather(
*[asyncio.to_thread(requete_bloquante, u) for u in urls]
)
for url, statut in resultats:
print(statut, url)
asyncio.run(main())requests reste du requests, mais les quatre appels partent en même temps. C'est le pont le plus rapide entre ton code actuel et le monde asynchrone.
Les trois réflexes à garder
Premier : async def ne fait rien tant que personne ne l'exécute, toujours via asyncio.run à la racine. Deuxième : n'await jamais en boucle si tu peux gather. Troisième : si une fonction bloque (SQL, HTTP synchrone, un gros fichier), enveloppe-la dans asyncio.to_thread plutôt que de réécrire tout ton projet.
asyncio ne rend pas Internet plus rapide, il supprime le temps où tu attends bêtement. Pour un script qui interroge un LLM, un lot d'images à traiter ou une cinquantaine d'endpoints à sonder, c'est souvent la différence entre un café et une pause déjeuner. Installe aiohttp, transforme ta boucle en gather, et mesure : le chrono te dira merci.
Auteur
Krugz
Développeur fullstack
Développeur fullstack qui tente, tant bien que mal, de suivre l'actualité de l'IA au milieu d'une marée d'informations imbuvables et des milliers d'outils et de modèles qui sortent chaque jour.
