Spaces:
Sleeping
Sleeping
File size: 6,940 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | # Pipeline Engine
A generic, composable step runner for ordered and parallel data processing.
---
## What is the Pipeline Engine?
The Pipeline Engine is a lightweight, domain-agnostic framework for composing processing steps into pipelines. It provides contract validation, immutable context passing, and built-in concurrency control — all in ~300 lines of pure Python with no external dependencies beyond the standard library.
Everything composes from three primitives:
**Sequential** — steps run one after another:
```mermaid
graph LR
A1[Step A] --> B1[Step B] --> C1[Step C]
```
**Branch** — fork, run in parallel, join:
```mermaid
graph LR
A2[Step A] --> B2[Step B] & C2[Step C] --> D2[Step D]
```
**Nesting** — a pipeline used as a step:
```mermaid
graph LR
A3[Step A] --> P3[[Inner Pipeline]] --> D3[Step D]
```
Steps declare what data they read and write. The pipeline validates ordering at construction time — before any data flows — so wiring errors surface immediately, not at runtime.
---
## Core Principles
- **Three primitives** — Sequential steps, parallel branches, and nested pipelines cover every composition pattern
- **Contracts** — Steps declare `requires` and `provides` fields; the pipeline validates ordering at construction time
- **Immutable context** — Steps receive a frozen context and return a new one via `.replace()`, making concurrent execution safe by default
- **Declared concurrency** — Parallelism is configured on the step (`max_workers`, `async_boundary`), not the pipeline
- **Per-sample error isolation** — One failing sample never blocks others; every sample produces a result
- **Observation hooks** — `PipelineHook` lets external code observe step transitions without modifying data flow (progress streaming, metrics, logging)
- **Cancellation** — `CancellationToken` stops a running pipeline between steps; `cancel_token_var` makes the token readable inside steps for intra-step cancellation
---
## Architecture at a Glance
```mermaid
classDiagram
class StepProtocol {
<<protocol>>
+requires: set[str]
+provides: set[str]
+__call__(ctx: StepContext) StepContext
}
class Pipeline {
+then(step) Pipeline
+branch(*pipelines) Pipeline
+run(samples) list~SampleResult~
+run_async(samples) list~SampleResult~
}
class Branch {
+merge: MergeStrategy
}
class YourStep {
+requires: set[str]
+provides: set[str]
+__call__(ctx) StepContext
}
class StepContext {
<<frozen dataclass>>
+sample: str
+metadata: MappingProxyType
+replace(**kw) StepContext
}
class SampleResult {
<<dataclass>>
+context: StepContext
+error: Exception?
+ok: bool
}
StepProtocol <|.. Pipeline : satisfies
StepProtocol <|.. Branch : satisfies
StepProtocol <|.. YourStep : satisfies
Pipeline *-- "1..*" StepProtocol : contains steps
Branch *-- "2..*" Pipeline : contains pipelines
StepProtocol ..> StepContext : receives & returns
Pipeline ..> SampleResult : produces
```
`Pipeline` and `Branch` both satisfy `StepProtocol` through structural typing — no inheritance required. This means a `Pipeline` can be used as a step inside another pipeline, and a `Branch` slots into any step position.
| Concept | What it is | Threading | Data flow |
|---------|-----------|-----------|-----------|
| **Step** | Single unit of work | Sync internally | Receives and returns `StepContext` |
| **Pipeline** | Ordered chain of steps | `workers=N` across samples | Passes `StepContext` step-to-step |
| **Branch** | Parallel fork/join | One thread per branch | Copies context in, merges outputs |
| **Nested Pipeline** | Pipeline used as a step | Inherits parent threading | Same `StepContext` flow |
---
## Async Boundary — Background Processing
One of the engine's key features is the **async boundary**: a way to split a pipeline into foreground (fast return) and background (fire-and-forget) stages.
```mermaid
graph LR
S1["Step A"] --> S2["Step B"] --> AB{{"async_boundary"}} --> S3["Step C<br/><small>background</small>"] --> S4["Step D<br/><small>background</small>"]
style S1 fill:#6366f1,stroke:#4f46e5,color:#fff
style S2 fill:#6366f1,stroke:#4f46e5,color:#fff
style AB fill:#f59e0b,stroke:#d97706,color:#000
style S3 fill:#3b82f6,stroke:#2563eb,color:#fff
style S4 fill:#3b82f6,stroke:#2563eb,color:#fff
```
Mark any step with `async_boundary = True` — the pipeline returns results immediately after the foreground steps, while everything from the boundary onward continues in background threads. Use `pipe.wait_for_background()` when you need the final results.
This is critical for pipelines where early steps produce user-facing output quickly but later steps (analysis, logging, scoring) are slow and don't need to block the caller. See [Execution Model](execution.md) for full details.
---
## When to Use
!!! tip "Good fit"
- Ordered multi-step processing with explicit data dependencies
- Parallel fork/join patterns (multiple independent operations on the same data)
- Fire-and-forget background processing with `async_boundary`
- Any pipeline where you want construction-time contract validation
!!! note "Not designed for"
- DAG scheduling with complex dependency graphs
- Distributed computing across multiple machines
- Stream processing with backpressure
- ETL pipelines requiring a data catalog
---
## Installation
The pipeline engine is included in the project with no extra dependencies:
```python
from pipeline import Pipeline, Branch, StepContext, MergeStrategy, PipelineHook, CancellationToken
```
!!! tip "Using the Pipeline Engine with ACE"
If you're building ACE pipelines, see [Composing Pipelines](../guides/composing-pipelines.md)
for ACE-specific steps and patterns. All pipeline classes are also importable
from `ace` directly: `from ace import Pipeline, Branch, ...`
---
## What's Next
- [**Quick Start**](quick-start.md) — Build and run your first pipeline in under 30 lines
- [**Core Concepts**](core-concepts.md) — Understand Step, Context, and the contract system
- [**Execution Model**](execution.md) — Three types of async, workers, and background processing
- [**Branching & Parallelism**](branching.md) — Parallel fork/join with merge strategies
- [**Error Handling**](error-handling.md) — Per-sample isolation, SampleResult, and error types
- [**Building Custom Steps**](custom-steps.md) — Create your own steps with dependency injection
- [**API Reference**](api-reference.md) — Complete signatures for all public classes
|