Stateful workers
One ML model per worker, loaded once
.class_task(MyClass, MyClass.method, args) instantiates MyClass once per worker. The instance persists across items — perfect for ML models, DB connections, accumulators.
Signature
Pipeline.class_task(class_constructor, class_method, class_args=None,
close_method=None, class_kwargs=None, count=1) -> Pipeline[OutPacket]from olympipe import Pipeline
from transformers import pipeline as hf_pipeline
class EmbeddingWorker:
def __init__(self, model_name: str):
# Loaded once at worker startup
self.model = hf_pipeline("feature-extraction", model=model_name)
def embed(self, text: str) -> list:
return self.model(text)[0][0] # CLS token
# 2 GPU workers, each loads the model once
embeddings = (
Pipeline(texts) # 50 000 texts
.batch(32)
.class_task(
EmbeddingWorker,
EmbeddingWorker.embed,
["sentence-transformers/all-MiniLM-L6-v2"],
count=2,
)
.wait_for_result()
)⚡ Performance
50 000 texts, HuggingFace embeddings
How it works
When the worker process starts, the constructor is called with args. All subsequent invocations reuse the same instance — model weights stay in RAM (or VRAM) without reloading.
class_kwargs and close_method
Full signature: .class_task(class_constructor, class_method, class_args=None, close_method=None, class_kwargs=None, count=1). class_args and class_kwargs are passed to the constructor in each worker. close_method is called once per worker at shutdown, to close a connection or release a model.
import os
import time
from olympipe import Pipeline
class Scaler:
def __init__(self, factor: int, offset: int = 0):
self.factor, self.offset = factor, offset
self.seen = 0
def apply(self, x: int) -> int:
time.sleep(0.05)
self.seen += 1
return x * self.factor + self.offset
def close(self) -> None: # called once per worker at shutdown
print(f"worker {os.getpid()} closed after {self.seen} items")
res = (
Pipeline(range(6))
.class_task(
Scaler,
Scaler.apply,
class_args=[10], # positional constructor arguments
class_kwargs={"offset": 1}, # keyword constructor arguments
close_method=Scaler.close,
count=2,
)
.wait_for_result()
)
print(sorted(res))worker 62409 closed after 3 items
worker 62411 closed after 3 items
[1, 11, 21, 31, 41, 51]Related examples
ML embedding generation
Generate embeddings for 50 000 texts with sentence-transformers.
LLM inference server
Serve a HuggingFace model as an HTTP API with Olympipe.
Parallel audio processing
Transcribe audio files with Whisper in parallel.