Patterns async avancés pour LangChain : Optimisation des performances et parallélisation
tl;dr:
L'async permet d'exécuter plusieurs opérations LLM en parallèle et de multiplier le throughput par 5-10x. Ce guide couvre async/await avec LangChain, concurrent execution, batch processing, streaming, backpressure, semaphores, queues, et profiling de performance. Essentiel pour systèmes haute performance.
Les applications LLM sont I/O-bound : elles passent la majorité du temps à attendre les réponses des APIs. L’async permet d’exécuter plusieurs requêtes en parallèle et de multiplier le throughput par 5-10x. Dans cet article, nous couvrons les patterns async avancés pour maximiser les performances de vos applications LangChain.
Pourquoi l’async est critique ?
Comparaison sync vs async
# ❌ Synchrone : LENT (10 requêtes × 2s = 20s)
import time
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
start = time.time()
results = []
for i in range(10):
result = llm.invoke(f"Question {i}") # Bloque pendant 2s
results.append(result)
print(f"Temps total: {time.time() - start:.1f}s") # ~20s
# ✅ Asynchrone : RAPIDE (10 requêtes en parallèle = 2s)
import asyncio
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
async def process_async():
start = time.time()
tasks = [
llm.ainvoke(f"Question {i}")
for i in range(10)
]
results = await asyncio.gather(*tasks) # Parallèle !
print(f"Temps total: {time.time() - start:.1f}s") # ~2s
return results
asyncio.run(process_async())
Gain : 10x plus rapide ! 🚀
Async/Await basics avec LangChain
Méthodes async natives
# chains/async_chain.py
import asyncio
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
class AsyncChain:
"""Chain avec support async natif"""
def __init__(self):
self.llm = ChatOpenAI(temperature=0)
self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")
self.chain = self.prompt | self.llm
async def ainvoke(self, query: str) -> str:
"""Invoke asynchrone (non-bloquant)"""
result = await self.chain.ainvoke({"query": query})
return result.content
async def abatch(self, queries: list[str]) -> list[str]:
"""Batch asynchrone (parallèle)"""
results = await self.chain.abatch([{"query": q} for q in queries])
return [r.content for r in results]
async def astream(self, query: str):
"""Stream asynchrone (tokens en temps réel)"""
async for chunk in self.chain.astream({"query": query}):
yield chunk.content
Utilisation
async def main():
chain = AsyncChain()
# Single invoke
response = await chain.ainvoke("Quelle est la capitale de la France ?")
print(response)
# Batch parallèle
queries = [
"Capitale de l'Allemagne ?",
"Capitale de l'Italie ?",
"Capitale de l'Espagne ?"
]
responses = await chain.abatch(queries)
for q, r in zip(queries, responses):
print(f"{q} → {r}")
# Streaming
async for chunk in chain.astream("Explique la photosynthèse"):
print(chunk, end="", flush=True)
# Exécuter
asyncio.run(main())
Concurrent execution avec asyncio.gather
Exécution parallèle de multiples chains
# processing/concurrent_executor.py
import asyncio
from typing import List, Dict
import time
class ConcurrentExecutor:
"""Exécute plusieurs chains en parallèle"""
def __init__(self, chain):
self.chain = chain
async def execute_parallel(
self,
queries: List[str],
max_concurrent: int = 10
) -> List[Dict]:
"""
Exécute queries en parallèle avec limite de concurrence
Args:
queries: Liste de queries
max_concurrent: Nombre max de requêtes simultanées
"""
# Semaphore pour limiter concurrence
semaphore = asyncio.Semaphore(max_concurrent)
async def process_with_limit(query: str, index: int) -> Dict:
async with semaphore:
start = time.time()
try:
result = await self.chain.ainvoke(query)
return {
"index": index,
"query": query,
"result": result,
"latency_ms": (time.time() - start) * 1000,
"success": True
}
except Exception as e:
return {
"index": index,
"query": query,
"error": str(e),
"success": False
}
# Créer toutes les tasks
tasks = [
process_with_limit(query, i)
for i, query in enumerate(queries)
]
# Exécuter en parallèle
results = await asyncio.gather(*tasks, return_exceptions=False)
return results
Utilisation
async def main():
chain = AsyncChain()
executor = ConcurrentExecutor(chain)
# 100 queries en parallèle (max 10 simultanées)
queries = [f"Question numéro {i}" for i in range(100)]
start = time.time()
results = await executor.execute_parallel(
queries,
max_concurrent=10
)
total_time = time.time() - start
# Statistiques
successful = sum(1 for r in results if r["success"])
avg_latency = sum(r["latency_ms"] for r in results if r["success"]) / successful
print(f"Total: {total_time:.1f}s")
print(f"Succès: {successful}/{len(queries)}")
print(f"Latence moyenne: {avg_latency:.0f}ms")
print(f"Throughput: {len(queries)/total_time:.1f} req/s")
asyncio.run(main())
Batch processing optimisé
Batch avec retry et backoff
# processing/batch_processor.py
import asyncio
from typing import List, Dict, Callable
import time
class AsyncBatchProcessor:
"""Traitement par batch avec retry et optimisations"""
def __init__(
self,
chain,
batch_size: int = 10,
max_retries: int = 3,
retry_delay: float = 1.0
):
self.chain = chain
self.batch_size = batch_size
self.max_retries = max_retries
self.retry_delay = retry_delay
async def process_batch(
self,
items: List[str],
callback: Callable = None
) -> List[Dict]:
"""
Process items en batches avec retry
Args:
items: Liste d'items à traiter
callback: Fonction appelée après chaque item
"""
results = []
# Découper en batches
for i in range(0, len(items), self.batch_size):
batch = items[i:i + self.batch_size]
batch_results = await self._process_single_batch(batch)
results.extend(batch_results)
# Callback optionnel
if callback:
callback(len(results), len(items))
return results
async def _process_single_batch(
self,
batch: List[str]
) -> List[Dict]:
"""Process un seul batch avec retry"""
tasks = [
self._process_with_retry(item, index)
for index, item in enumerate(batch)
]
return await asyncio.gather(*tasks)
async def _process_with_retry(
self,
item: str,
index: int
) -> Dict:
"""Process un item avec retry"""
for attempt in range(self.max_retries + 1):
try:
start = time.time()
result = await self.chain.ainvoke(item)
return {
"index": index,
"item": item,
"result": result,
"latency_ms": (time.time() - start) * 1000,
"attempts": attempt + 1,
"success": True
}
except Exception as e:
if attempt == self.max_retries:
return {
"index": index,
"item": item,
"error": str(e),
"attempts": attempt + 1,
"success": False
}
# Exponential backoff
await asyncio.sleep(self.retry_delay * (2 ** attempt))
Utilisation avec progress callback
async def main():
chain = AsyncChain()
processor = AsyncBatchProcessor(
chain,
batch_size=10,
max_retries=3
)
items = [f"Item {i}" for i in range(100)]
# Callback pour progress
def progress_callback(completed, total):
progress = (completed / total) * 100
print(f"Progrès: {completed}/{total} ({progress:.1f}%)", end="\r")
results = await processor.process_batch(items, callback=progress_callback)
print(f"\nTraité: {len(results)} items")
print(f"Succès: {sum(1 for r in results if r['success'])}")
asyncio.run(main())
Streaming async avancé
Stream processing avec buffering
# streaming/async_streamer.py
import asyncio
from typing import AsyncIterator
from collections import deque
class AsyncStreamer:
"""Streaming avec buffering intelligent"""
def __init__(self, chain, buffer_size: int = 100):
self.chain = chain
self.buffer_size = buffer_size
async def stream_with_buffer(
self,
query: str
) -> AsyncIterator[str]:
"""Stream avec buffer pour smoothing"""
buffer = deque(maxlen=self.buffer_size)
async for chunk in self.chain.astream({"query": query}):
buffer.append(chunk.content)
# Yield par petits groupes
if len(buffer) >= 10:
yield "".join(buffer)
buffer.clear()
# Yield le reste du buffer
if buffer:
yield "".join(buffer)
async def stream_multiple(
self,
queries: List[str]
) -> AsyncIterator[Dict]:
"""Stream plusieurs queries en parallèle"""
tasks = []
for i, query in enumerate(queries):
task = self._stream_single(query, i)
tasks.append(task)
# Yield dès qu'une query produit un chunk
for task in asyncio.as_completed(tasks):
result = await task
yield result
async def _stream_single(
self,
query: str,
query_id: int
) -> Dict:
"""Stream une seule query"""
chunks = []
async for chunk in self.chain.astream({"query": query}):
chunks.append(chunk.content)
return {
"query_id": query_id,
"query": query,
"response": "".join(chunks)
}
Backpressure et rate limiting
Contrôle du débit avec asyncio.Queue
# processing/rate_limited_processor.py
import asyncio
from asyncio import Queue
import time
class RateLimitedProcessor:
"""Processor avec rate limiting et backpressure"""
def __init__(
self,
chain,
max_requests_per_second: int = 10,
queue_size: int = 1000
):
self.chain = chain
self.max_rps = max_requests_per_second
self.queue = Queue(maxsize=queue_size)
self.results = []
async def add_request(self, query: str):
"""Ajoute une requête à la queue (peut bloquer si pleine)"""
await self.queue.put(query)
async def process_queue(self):
"""Process la queue avec rate limiting"""
interval = 1.0 / self.max_rps # Intervalle entre requêtes
while True:
try:
# Récupérer la prochaine requête
query = await asyncio.wait_for(
self.queue.get(),
timeout=1.0
)
# Process
start = time.time()
result = await self.chain.ainvoke(query)
self.results.append({
"query": query,
"result": result,
"latency_ms": (time.time() - start) * 1000
})
# Rate limiting
elapsed = time.time() - start
if elapsed < interval:
await asyncio.sleep(interval - elapsed)
except asyncio.TimeoutError:
# Queue vide, continuer
continue
except Exception as e:
print(f"Erreur: {e}")
continue
async def run(self, queries: List[str]):
"""Lance le processing avec rate limiting"""
# Démarrer le processor
processor_task = asyncio.create_task(self.process_queue())
# Ajouter toutes les queries
for query in queries:
await self.add_request(query)
# Attendre que la queue soit vide
await self.queue.join()
# Arrêter le processor
processor_task.cancel()
return self.results
Patterns avancés
Worker pool pattern
# processing/worker_pool.py
import asyncio
from typing import List, Callable
class WorkerPool:
"""Pool de workers async"""
def __init__(
self,
worker_func: Callable,
num_workers: int = 5
):
self.worker_func = worker_func
self.num_workers = num_workers
self.queue = asyncio.Queue()
self.results = []
async def worker(self, worker_id: int):
"""Worker qui process les items de la queue"""
while True:
try:
item = await self.queue.get()
if item is None: # Signal d'arrêt
break
result = await self.worker_func(item)
self.results.append(result)
self.queue.task_done()
except Exception as e:
print(f"Worker {worker_id} error: {e}")
self.queue.task_done()
async def process(self, items: List) -> List:
"""Process items avec le pool de workers"""
# Démarrer les workers
workers = [
asyncio.create_task(self.worker(i))
for i in range(self.num_workers)
]
# Ajouter les items à la queue
for item in items:
await self.queue.put(item)
# Attendre que tout soit traité
await self.queue.join()
# Arrêter les workers
for _ in range(self.num_workers):
await self.queue.put(None)
await asyncio.gather(*workers)
return self.results
Pipeline pattern
# processing/pipeline.py
import asyncio
from typing import Callable, List
class AsyncPipeline:
"""Pipeline de transformations async"""
def __init__(self, stages: List[Callable]):
self.stages = stages
async def process_item(self, item):
"""Process un item à travers tous les stages"""
result = item
for stage in self.stages:
result = await stage(result)
return result
async def process_batch(self, items: List) -> List:
"""Process un batch en parallèle"""
tasks = [self.process_item(item) for item in items]
return await asyncio.gather(*tasks)
# Utilisation
async def stage1(item):
# Enrichissement
await asyncio.sleep(0.1)
return f"Stage1({item})"
async def stage2(item):
# Validation
await asyncio.sleep(0.1)
return f"Stage2({item})"
async def stage3(item):
# LLM processing
await asyncio.sleep(0.5)
return f"Stage3({item})"
pipeline = AsyncPipeline([stage1, stage2, stage3])
results = await pipeline.process_batch(["item1", "item2", "item3"])
Performance profiling
Mesurer les performances
# profiling/async_profiler.py
import asyncio
import time
from typing import Dict, List
from dataclasses import dataclass
from statistics import mean, stdev
@dataclass
class ProfileResult:
total_time: float
num_requests: int
successful: int
failed: int
avg_latency: float
std_latency: float
min_latency: float
max_latency: float
throughput: float # requests/second
class AsyncProfiler:
"""Profile les performances async"""
@staticmethod
async def profile_chain(
chain,
queries: List[str],
max_concurrent: int = 10
) -> ProfileResult:
"""Profile une chain avec différentes charges"""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def timed_invoke(query: str) -> Dict:
async with semaphore:
start = time.time()
try:
result = await chain.ainvoke(query)
latency = time.time() - start
return {
"success": True,
"latency": latency
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
# Exécuter
start = time.time()
tasks = [timed_invoke(q) for q in queries]
results = await asyncio.gather(*tasks)
total_time = time.time() - start
# Analyser
successful = [r for r in results if r["success"]]
failed = [r for r in results if not r["success"]]
latencies = [r["latency"] for r in successful]
return ProfileResult(
total_time=total_time,
num_requests=len(queries),
successful=len(successful),
failed=len(failed),
avg_latency=mean(latencies) if latencies else 0,
std_latency=stdev(latencies) if len(latencies) > 1 else 0,
min_latency=min(latencies) if latencies else 0,
max_latency=max(latencies) if latencies else 0,
throughput=len(queries) / total_time
)
@staticmethod
async def benchmark_concurrency(
chain,
query: str,
concurrency_levels: List[int] = [1, 5, 10, 20, 50]
) -> Dict[int, ProfileResult]:
"""Benchmark différents niveaux de concurrence"""
results = {}
for concurrency in concurrency_levels:
print(f"Testing concurrency: {concurrency}...")
# Créer N requêtes identiques
queries = [query] * 100
result = await AsyncProfiler.profile_chain(
chain,
queries,
max_concurrent=concurrency
)
results[concurrency] = result
print(f" Throughput: {result.throughput:.1f} req/s")
print(f" Latence: {result.avg_latency*1000:.0f}ms")
return results
Utilisation
async def main():
chain = AsyncChain()
# Profile simple
queries = [f"Question {i}" for i in range(100)]
result = await AsyncProfiler.profile_chain(
chain,
queries,
max_concurrent=10
)
print(f"Total: {result.total_time:.1f}s")
print(f"Throughput: {result.throughput:.1f} req/s")
print(f"Latence moyenne: {result.avg_latency*1000:.0f}ms ± {result.std_latency*1000:.0f}ms")
print(f"Min/Max: {result.min_latency*1000:.0f}ms / {result.max_latency*1000:.0f}ms")
# Benchmark concurrence
benchmark_results = await AsyncProfiler.benchmark_concurrency(
chain,
"Test query",
concurrency_levels=[1, 5, 10, 20, 50]
)
# Afficher courbe
print("\nConcurrence vs Throughput:")
for concurrency, result in benchmark_results.items():
print(f"{concurrency:3d} → {result.throughput:6.1f} req/s")
asyncio.run(main())
Best practices
Toujours utiliser async pour I/O
# ✅ BON : Async pour appels LLM
async def process_queries(queries):
tasks = [chain.ainvoke(q) for q in queries]
return await asyncio.gather(*tasks)
# ❌ MAUVAIS : Sync (bloque le thread)
def process_queries_sync(queries):
return [chain.invoke(q) for q in queries]
Limiter la concurrence
# ✅ BON : Semaphore pour limiter
semaphore = asyncio.Semaphore(10) # Max 10 simultanés
async def limited_invoke(query):
async with semaphore:
return await chain.ainvoke(query)
# ❌ MAUVAIS : Concurrence illimitée (peut surcharger l'API)
tasks = [chain.ainvoke(q) for q in huge_list] # 10 000 en parallèle !
Gérer les timeouts
# ✅ BON : Timeout global
try:
result = await asyncio.wait_for(
chain.ainvoke(query),
timeout=30.0
)
except asyncio.TimeoutError:
# Gérer le timeout
# ❌ MAUVAIS : Pas de timeout (peut bloquer indéfiniment)
result = await chain.ainvoke(query)
Utiliser return_exceptions dans gather
# ✅ BON : Continue même si certaines tasks échouent
results = await asyncio.gather(
*tasks,
return_exceptions=True
)
# Filtrer les erreurs
successful = [r for r in results if not isinstance(r, Exception)]
errors = [r for r in results if isinstance(r, Exception)]
# ❌ MAUVAIS : S'arrête à la première erreur
results = await asyncio.gather(*tasks) # Lève exception
Checklist performance
Avant le déploiement
- Async activé : Toutes les opérations I/O sont async
- Concurrence limitée : Semaphore ou rate limiter
- Batch processing : Requêtes groupées quand possible
- Timeouts : Définis à tous les niveaux
- Error handling : return_exceptions=True dans gather
- Profiling : Performance mesurée avec différentes charges
- Optimal concurrency : Niveau de concurrence testé
- Monitoring : Métriques de throughput et latence
En production
- Throughput : > 50 req/s (ajuster selon besoin)
- Latence P95 : < 3s
- Error rate : < 1%
- Memory usage : Stable (pas de leak)
- CPU usage : < 70% en moyenne
- Scaling : Horizontal scaling testé
Ressources complémentaires
Documentation
Articles connexes
- Erreurs & Résilience - Error handling async
- Testing - Tests async
- LangSmith - Monitoring performance
Conclusion
L’async est essentiel pour la performance des applications LLM. Les gains sont massifs :
- 5-10x throughput plus élevé
- Meilleure utilisation des ressources
- Expérience utilisateur améliorée
Les patterns clés :
- asyncio.gather pour parallélisation
- Semaphores pour limiter concurrence
- Queues pour backpressure
- Worker pools pour processing distribué
- Profiling pour optimisation
Avec ces patterns, vous construisez des systèmes haute performance capables de gérer des milliers de requêtes par seconde.
Prochaine étape : Optimiser les coûts avec tracking et budgets !
Navigation
← Retour à la Formation LangChain | Module suivant : Gestion et Optimisation des Coûts →