Skip to content

pipeline

pipeline

Attributes

__doctitle__ module-attribute 

__doctitle__ = 'Pipeline (Async)'

__all__ module-attribute 

__all__ = ['Pipeline', 'classification', 'detection', 'segmentation']

Classes

Pipeline 

Pipeline(runtime: Runtime[P, R], batch_strategy: BatchStrategy[P, R] | None = None)

Bases: ABC, Generic[P, R, O]

Abstract inference pipeline (async version).

A pipeline combines: - Preprocessing: Convert raw input to model-ready format - Inference: Run model inference via runtime - Postprocessing: Convert raw output to structured result

Example 
pipeline = ClassificationPipeline(runtime=runtime)
async with pipeline.serve():
    result = await pipeline(image)
    print(
        f"Class: {result.class_name}, Confidence: {result.confidence}"
    )

Initialize pipeline.

Parameters:

Name Type Description Default
runtime Runtime[P, R]

Inference runtime.

 required
batch_strategy BatchStrategy[P, R] | None

Optional batching strategy for improved throughput.

 None
Source code in inferflow/asyncio/pipeline/__init__.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    runtime: Runtime[P, R],
    batch_strategy: BatchStrategy[P, R] | None = None,
):
    """Initialize pipeline.

    Args:
        runtime: Inference runtime.
        batch_strategy: Optional batching strategy for improved throughput.
    """
    self.runtime = runtime
    self.batch_strategy = batch_strategy
Attributes
runtime instance-attribute 
runtime = runtime
batch_strategy instance-attribute 
batch_strategy = batch_strategy
Functions
preprocess abstractmethod async 
preprocess(input: ImageInput) -> P

Preprocess raw input into model-ready format.

Parameters:

Name Type Description Default
input ImageInput

Raw input (image bytes, numpy array, PIL Image, etc.)

 required

Returns:

Type Description
P

Preprocessed input ready for inference.
Source code in inferflow/asyncio/pipeline/__init__.py 
53
54
55
56
57
58
59
60
61
62
@abc.abstractmethod
async def preprocess(self, input: ImageInput) -> P:
    """Preprocess raw input into model-ready format.

    Args:
        input: Raw input (image bytes, numpy array, PIL Image, etc.)

    Returns:
        Preprocessed input ready for inference.
    """
postprocess abstractmethod async 
postprocess(raw: R) -> O

Postprocess raw model output into structured result.

Parameters:

Name Type Description Default
raw R

Raw output from model inference.

 required

Returns:

Type Description
O

Structured output (classification result, detections, etc.)
Source code in inferflow/asyncio/pipeline/__init__.py 
64
65
66
67
68
69
70
71
72
73
@abc.abstractmethod
async def postprocess(self, raw: R) -> O:
    """Postprocess raw model output into structured result.

    Args:
        raw: Raw output from model inference.

    Returns:
        Structured output (classification result, detections, etc.)
    """
infer async 
infer(preprocessed: P) -> R

Run inference on preprocessed input.

This method automatically uses batching if a batch strategy is configured.

Parameters:

Name Type Description Default
preprocessed P

Preprocessed input.

 required

Returns:

Type Description
R

Raw inference result.
Source code in inferflow/asyncio/pipeline/__init__.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def infer(self, preprocessed: P) -> R:
    """Run inference on preprocessed input.

    This method automatically uses batching if a batch strategy is configured.

    Args:
        preprocessed: Preprocessed input.

    Returns:
        Raw inference result.
    """
    if self.batch_strategy:
        return await self.batch_strategy.submit(preprocessed)
    return await self.runtime.infer(preprocessed)
__call__ async 
__call__(input: ImageInput) -> O

End-to-end inference.

Parameters:

Name Type Description Default
input ImageInput

Raw input.

 required

Returns:

Type Description
O

Structured output.
Example 
result = await pipeline(image_bytes)
Source code in inferflow/asyncio/pipeline/__init__.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def __call__(self, input: ImageInput) -> O:
    """End-to-end inference.

    Args:
        input: Raw input.

    Returns:
        Structured output.

    Example:
        ```python
        result = await pipeline(image_bytes)
        ```
    """
    preprocessed = await self.preprocess(input)
    raw = await self.infer(preprocessed)
    return await self.postprocess(raw)
serve async 
serve() -> AsyncIterator[Self]

Start serving pipeline with automatic lifecycle management.

This method
  • Loads the runtime
  • Starts batch processing (if enabled)
  • Yields the pipeline for inference
  • Cleans up resources on exit
Example 
async with pipeline.serve():
    result = await pipeline(image)
Source code in inferflow/asyncio/pipeline/__init__.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@contextlib.asynccontextmanager
async def serve(self) -> t.AsyncIterator[t.Self]:
    """Start serving pipeline with automatic lifecycle management.

    This method:
        - Loads the runtime
        - Starts batch processing (if enabled)
        - Yields the pipeline for inference
        - Cleans up resources on exit

    Example:
        ```python
        async with pipeline.serve():
            result = await pipeline(image)
        ```
    """
    async with self.runtime.context():
        if self.batch_strategy:
            await self.batch_strategy.start(self.runtime)

        try:
            yield self
        finally:
            if self.batch_strategy:
                await self.batch_strategy.stop()

Functions

__getattr__ 

__getattr__(name: str) -> Any
Source code in inferflow/asyncio/pipeline/__init__.py 
138
139
140
141
def __getattr__(name: str) -> t.Any:
    if name in __all__:
        return importlib.import_module("." + name, __name__)
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

Submodules