Split & Gather
Branchez votre pipeline en flux indépendants, fusionnez-les après
.split(fn, n=2) route chaque item vers 0 à N branches selon la valeur retournée par fn. .gather(*pipes) fusionne N pipelines en un seul flux. Idéal pour traiter différentes catégories d'items avec des logiques distinctes.
Signature
Pipeline.split(split_function, n=None) -> tuple[Pipeline, ...]
Pipeline.gather(*others: Pipeline) -> Pipelinefrom typing import Optional, Tuple
from olympipe import Pipeline
def route_by_type(item: dict) -> Tuple[Optional[dict], Optional[dict]]:
if item["type"] == "image":
return (item, None) # → branche images
return (None, item) # → branche textes
images, texts = Pipeline(mixed_dataset).split(route_by_type, n=2)
results = (
images.task(process_image, count=4) # pipeline images
.gather(texts.task(process_text, count=2)) # + pipeline textes
.wait_for_result()
)⚡ Performance
5 000 items mixtes, traitement différencié
Comment ça fonctionne
Chaque branche est un Pipeline indépendant avec sa propre file. None dans la valeur de retour signifie « ne pas envoyer sur cette branche ». gather branche chaque pipeline sur une file de sortie commune : les items sont fusionnés au fil de leur arrivée, sans ordre garanti.
split sans n, et 3 ou 4 branches
Signature : .split(split_function, n=None). Si n est omis, il est déduit de l'annotation de retour (Tuple[...] à 2, 3 ou 4 éléments, ou plus) ; sans annotation, split lève ValueError. La fonction renvoie un tuple de n valeurs : chaque valeur non None part sur la branche correspondante.
from typing import Optional, Tuple
from olympipe import Pipeline
Route = Tuple[Optional[int], Optional[int], Optional[int]]
def by_remainder(x: int) -> Route:
out: list = [None, None, None]
out[x % 3] = x
return tuple(out)
r0, r1, r2 = Pipeline(range(9)).split(by_remainder) # 3 branches, n déduit
print(r0.wait_for_results([r1, r2]))
def no_annotation(x):
return (x, None)
try:
Pipeline(range(3)).split(no_annotation)
except ValueError as e:
print("ValueError:", e)[[0, 3, 6], [1, 4, 7], [2, 5, 8]]
ValueError: Cannot infer number of outputs for 'no_annotation'. Add a return annotation (e.g. Tuple[A, B]) or pass n= explicitly.