Spaces:
Sleeping
Sleeping
File size: 9,493 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 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 | # Core Concepts
The pipeline engine is built on four foundational concepts: the **Step protocol**, the **StepContext**, the **contract system**, and the **Pipeline** compositor. Understanding these gives you the mental model for everything else.
---
## StepProtocol
A Step is any Python object that satisfies the `StepProtocol` — a structural (duck-typed) interface. No base class is required.
```python
from collections.abc import Set as AbstractSet
from typing import Protocol, runtime_checkable
@runtime_checkable
class StepProtocol(Protocol):
requires: AbstractSet[str] # metadata keys this step reads
provides: AbstractSet[str] # metadata keys this step writes
def __call__(self, ctx: StepContext) -> StepContext: ...
```
Any object with `requires`, `provides`, and a `__call__` method is a valid step:
```python
class Tokenize:
requires = frozenset() # no dependencies
provides = frozenset({"tokens", "word_count"})
def __call__(self, ctx: StepContext) -> StepContext:
tokens = str(ctx.sample).split()
return ctx.replace(
metadata=MappingProxyType({
**ctx.metadata,
"tokens": tokens,
"word_count": len(tokens),
})
)
```
Key details:
- **`AbstractSet[str]`** accepts both `set` and `frozenset`. Steps can use plain set literals — the pipeline normalizes them to `frozenset` at construction time.
- **`@runtime_checkable`** lets the pipeline use `isinstance(step, StepProtocol)` at construction time to catch missing attributes early, rather than failing at call time.
- **`Pipeline` and `Branch`** both satisfy this protocol, so they can be nested wherever a step is expected.
---
## StepContext
`StepContext` is the data carrier passed from step to step. It is a **frozen dataclass** — steps never mutate the incoming context.
```python
from dataclasses import dataclass, field
from types import MappingProxyType
@dataclass(frozen=True)
class StepContext:
sample: Any = None
metadata: MappingProxyType = field(
default_factory=lambda: MappingProxyType({})
)
def replace(self, **changes) -> "StepContext":
return dataclasses.replace(self, **changes)
```
The engine only reads `sample` and `metadata`. All domain-specific fields are added by subclassing.
### The `.replace()` pattern
Steps create new contexts — they never mutate the incoming one:
```python
def __call__(self, ctx: StepContext) -> StepContext:
result = process(ctx.sample)
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "result": result})
)
```
This is the only way to "modify" a context. `frozen=True` makes mutation a hard error at runtime rather than a subtle bug.
### Metadata auto-coercion
If a caller passes a plain `dict` as metadata, `StepContext.__post_init__` automatically wraps it in `MappingProxyType`, ensuring mutation is always a runtime error:
```python
# Both of these produce identical immutable metadata:
ctx = StepContext(sample="hello", metadata={"key": "value"})
ctx = StepContext(sample="hello", metadata=MappingProxyType({"key": "value"}))
```
### Subclassing for domain fields
Applications subclass `StepContext` to add named fields for concepts shared across their pipelines:
```python
@dataclass(frozen=True)
class MLContext(StepContext):
# Shared configuration
model_config: dict | None = None
# Produced by steps (None until the providing step runs)
predictions: list | None = None
scores: dict | None = None
report: str | None = None
```
Use **named fields** for data shared across multiple steps in the pipeline. Use **`metadata`** for integration-specific or step-specific transient data that doesn't warrant a dedicated field.
!!! tip "When to use which"
- Named field: `predictions`, `scores` — shared by multiple steps, type-checkable
- Metadata: `metadata["debug_log"]`, `metadata["cache_key"]` — step-specific, doesn't pollute the class
### Why immutability?
- **Branch safety** — All branches receive the same frozen context. No deep copy is needed since no branch can mutate what it receives.
- **Thread safety** — Steps running concurrently (via `workers` or `Branch`) can safely share context objects.
- **Debugging** — Each step returns a new context, creating a clear trace of data transformations.
---
## Contracts: requires and provides
Every step declares:
- **`requires`** — the set of field names it reads from the context
- **`provides`** — the set of field names it writes to the context
The pipeline validates these at construction time.
### How validation works
When you build a pipeline with `.then()`, the engine checks step ordering immediately:
```python
pipe = (
Pipeline()
.then(Tokenize()) # provides: {"tokens", "word_count"}
.then(Uppercase()) # requires: {"tokens"} ✓ — Tokenize provides it
)
```
If a step requires a field that a **later** step provides, the pipeline raises `PipelineOrderError`:
```python
# This raises PipelineOrderError at construction time:
pipe = Pipeline().then(Uppercase()).then(Tokenize())
# ↑ Uppercase requires "tokens", but Tokenize (which provides it) comes after
```
### External inputs vs internal dependencies
Fields not produced by any step in the pipeline are treated as **external inputs** — they must be present in the initial `StepContext` passed to `run()`. These do not trigger ordering errors:
```python
class ScoreStep:
requires = frozenset({"predictions"}) # external input
provides = frozenset({"scores"})
# No error — "predictions" is expected to come from the initial context
pipe = Pipeline().then(ScoreStep())
```
### Contract inference for nested pipelines
When a `Pipeline` is used as a step inside another pipeline, its `requires` and `provides` are computed automatically from its inner steps:
```python
inner = Pipeline().then(Tokenize()).then(Uppercase())
# Inferred automatically:
# inner.requires = frozenset() — Tokenize needs nothing external
# inner.provides = frozenset({"tokens", "word_count", "upper_tokens"})
outer = Pipeline().then(inner).then(Summarize())
# Summarize's requirements validated against inner.provides
```
The inference algorithm:
1. Walk steps in order, tracking what has been provided so far
2. `requires` = fields needed by steps that no earlier step provides (external dependencies)
3. `provides` = union of all fields any step writes
!!! warning "All Branch children always run"
The contract system assumes all `Branch` children execute. There is no concept of conditional branches where only some children run — all branches always run. If a branch provides a field that a later step requires, validation passes; if that branch were to not run, the pipeline would fail at runtime.
---
## Pipeline
A `Pipeline` is an ordered list of steps that runs sequentially for a single input. It satisfies the `StepProtocol`, so it can be nested inside other pipelines.
### Building a pipeline
Two equivalent forms:
=== "Fluent builder (preferred)"
```python
pipe = (
Pipeline()
.then(Tokenize())
.then(Uppercase())
.then(Summarize())
)
```
=== "Constructor list"
```python
pipe = Pipeline([
Tokenize(),
Uppercase(),
Summarize(),
])
```
Both validate step ordering at construction time. The fluent builder validates **after each `.then()` call**, giving precise error messages about which step caused the violation.
### The `.branch()` shorthand
Instead of manually creating a `Branch`, use the fluent shorthand:
```python
pipe = (
Pipeline()
.then(Tokenize())
.branch(
Pipeline().then(Uppercase()),
Pipeline().then(Reverse()),
merge=MergeStrategy.RAISE_ON_CONFLICT,
)
.then(Summarize())
)
```
This is equivalent to `.then(Branch(...))`.
### Nesting
A pipeline used as a step is a black box — the outer pipeline sees only its aggregated `requires` and `provides`:
```python
preprocessing = Pipeline().then(Tokenize()).then(Uppercase())
postprocessing = Pipeline().then(Summarize()).then(FormatStep())
full = Pipeline().then(preprocessing).then(postprocessing)
```
!!! note "Inner pipeline as a fan-out step"
A step receives one context and must return one context — but nothing prevents it from internally expanding to multiple sub-inputs:
```python
class MultiSearchStep:
requires = frozenset()
provides = frozenset({"search_results"})
def __call__(self, ctx: StepContext) -> StepContext:
queries = generate_queries(ctx.sample)
sub_ctxs = [StepContext(sample=q) for q in queries]
sub_pipe = Pipeline().then(FetchStep())
results = sub_pipe.run(sub_ctxs, workers=len(queries))
merged = merge_results(results)
return ctx.replace(
metadata=MappingProxyType({**ctx.metadata, "search_results": merged})
)
```
From the outer pipeline's perspective, `MultiSearchStep` is a single step. The fan-out is an internal implementation detail.
|