Batching
Regroupez les items pour saturer votre GPU ou votre BDD
.batch(n) accumule n items avant de les émettre en liste. .temporal_batch(s) regroupe tout ce qui arrive pendant s secondes — parfait pour les flux temps-réel (audio, capteurs, WebSocket).
Signature
Pipeline.batch(batch_size=2, keep_incomplete_batch=True) -> Pipeline[list[InPacket]]
Pipeline.temporal_batch(time_interval: float) -> Pipeline[list[InPacket]]python
from olympipe import Pipeline
import time
# Batch fixe : regrouper 64 items avant traitement
results = (
Pipeline(image_paths) # 10 000 images
.task(load_image, count=4) # chargement parallèle
.batch(64) # groupe en batches
.task(model.predict) # inférence GPU
.wait_for_result()
)
# Batch temporel : fenêtre glissante de 500ms
results = (
Pipeline(sensor_stream)
.task(parse_frame)
.temporal_batch(0.5) # tout ce qui arrive en 500ms
.task(aggregate)
.wait_for_result()
)⚡ Performance
10 000 images, inférence GPU
Sans batch 50.0s
batch=16 12.0s
batch=64 3.2s
batch=128 2.1s
🚀 23.8× plus rapide
Comment ça fonctionne
Sans batch, chaque item déclenche un aller-retour GPU séparé. Avec un batch de 64, un seul appel traite 64 images — la latence GPU fixe est amortie sur 64 fois plus de travail.
keep_incomplete_batch
Signature : .batch(batch_size=2, keep_incomplete_batch=True). Par défaut, le dernier batch incomplet est émis à la fin du flux ; avec keep_incomplete_batch=False, il est abandonné (utile quand l'étape suivante exige une taille fixe).
python
from olympipe import Pipeline
print(Pipeline(range(7)).batch(3).wait_for_result())
print(Pipeline(range(7)).batch(3, keep_incomplete_batch=False).wait_for_result())Sortie
[[0, 1, 2], [3, 4, 5], [6]]
[[0, 1, 2], [3, 4, 5]]