Python's Iterator Protocol Applied to Streaming ML Inference

Five stages, five generators, one Pipeline — and the audit emits at the speed of the stream.

§ IFrame

Today's Ops lesson named three multi-agent orchestration patterns: Pipeline, Fan-Out, Specialist Team. This Dev lesson takes the Pipeline pattern and hands it its Python-language realization.

Python's iterator protocol is the language-level mechanism for sequential, single-direction, stage-to-stage data flow. It is also the language-level mechanism for the Pipeline pattern. The mapping is not metaphorical; it is structural. When an engineer writes a chain of generators in Python, they are writing a Pipeline in the exact shape this morning's Ops lesson described: stages, schemas at the seams, schema-check audit, principal-traceable chain of grants if logging is added at each yield point.

Streaming ML inference is the canonical use case where this mapping pays off. The model serves predictions on a stream of incoming requests; the pre-processor reads the request stream, transforms each item, and yields features; the inference engine consumes features and yields raw outputs; the post-processor consumes raw outputs and yields final responses; the response writer consumes final responses and writes to the response channel. Five stages, five generators, one Pipeline. The whole stack runs without ever buffering the full intermediate state, which is the architectural payoff this lesson is named for.

§ IILanguage Idiom

The iterator protocol in Python is a two-method contract. An iterator is any object that implements both methods.

class Iterator:
    def __iter__(self):
        return self

    def __next__(self):
        ...
        raise StopIteration

__iter__ returns the iterator itself, which makes the iterator usable in a for loop. __next__ returns the next element; when there are no more elements, it raises StopIteration, which signals end-of-stream to the consumer.

Generators are syntactic sugar over the iterator protocol. A function that contains a yield statement is automatically a generator function; calling it returns a generator object that implements both __iter__ and __next__ for free. The function's local state, including the position in the body, is preserved between calls to __next__, which makes generators well-suited to streaming work.

def integers_from(start):
    n = start
    while True:
        yield n
        n += 1

This generator function produces an unbounded stream of integers starting at start. The while True loop never returns; the yield statement returns control to the caller after each value. The caller pulls one value at a time using next(gen) or by iterating with for.

Generator expressions are the comprehension form of generators. The syntax mirrors list comprehensions but with parentheses instead of brackets, and the result is a generator object that produces values lazily rather than a list that holds them all in memory.

squares = (n * n for n in integers_from(1))

squares is a generator that yields 1, 4, 9, 16, ..., on demand, one at a time, forever. No memory is consumed for the stream beyond the single integer being yielded and the small fixed cost of the generator frame.

Async generators extend the protocol to asynchronous contexts. An async def function containing yield is an asynchronous generator; the consumer must use async for to iterate, and the iteration can await between elements. This matters for streaming ML inference because the I/O at most pipeline seams (model invocation, network reads, disk writes) is naturally async-shaped.

async def request_stream(connection):
    while True:
        item = await connection.read_request()
        if item is None:
            return
        yield item

Note the absence of inline comments throughout these blocks. The protocol is explained in the prose above and below; the code carries no commentary. This discipline forces the prose to do the explanatory work, which keeps the prose honest about whether the explanation is sufficient.

§ IIICode Worked Example

A streaming ML inference Pipeline built from generators. Five stages, each a generator function, chained at module top-level into the runnable inference loop.

The first stage reads the incoming request stream and yields parsed request objects.

async def parse_requests(connection):
    async for raw in request_stream(connection):
        parsed = Request.from_json(raw)
        yield parsed

The second stage consumes parsed requests and yields feature vectors.

async def featurize(request_iter, feature_store):
    async for req in request_iter:
        base_features = feature_store.lookup(req.entity_id)
        live_features = req.live_signals
        feature_vec = combine(base_features, live_features)
        yield (req.request_id, feature_vec)

The third stage consumes feature vectors and yields raw model outputs.

async def infer(feature_iter, model_client):
    async for request_id, feature_vec in feature_iter:
        raw_output = await model_client.predict(feature_vec)
        yield (request_id, raw_output)

The fourth stage consumes raw outputs and yields final responses.

async def post_process(output_iter, post_processor):
    async for request_id, raw in output_iter:
        final = post_processor.apply(raw)
        yield (request_id, final)

The fifth stage consumes final responses and writes them to the response channel.

async def write_responses(response_iter, connection):
    async for request_id, final in response_iter:
        await connection.write_response(request_id, final)

The top-level driver wires the five stages into a single chain.

async def run_pipeline(connection, feature_store, model_client, post_processor):
    requests = parse_requests(connection)
    features = featurize(requests, feature_store)
    raw_outputs = infer(features, model_client)
    final_responses = post_process(raw_outputs, post_processor)
    await write_responses(final_responses, connection)

The whole inference pipeline is fifteen lines plus the driver. Each stage is independently testable, independently profilable, independently swappable. The data flows through the chain without ever being buffered as a list; at any moment, the system holds at most one request per stage, which is five requests in flight total. Memory consumption is constant in the size of the request stream.

The audit-trail discipline asked for by the Ops lesson is a logging hook at every yield site. Adding it does not require restructuring the pipeline; it requires wrapping each generator in a logging decorator or inserting a logger.info(...) call before each yield. The chain remains intact; only the side-effect surface expands.

async def featurize_logged(request_iter, feature_store, logger):
    async for req in request_iter:
        base_features = feature_store.lookup(req.entity_id)
        live_features = req.live_signals
        feature_vec = combine(base_features, live_features)
        logger.info("featurized", request_id=req.request_id)
        yield (req.request_id, feature_vec)

§ IVConnection to Today's Ops Lesson

The Pipeline pattern named in the Ops lesson maps directly onto the generator chain above. Five stages in the Ops lesson; five generators in the Dev lesson. The Ops lesson's schema check between stages is the type contract on each generator's yielded value; for example, the featurize generator yields (request_id: str, feature_vec: np.ndarray). The Ops lesson's re-run from the failing stage is, in Python, a matter of restarting the failing generator and reattaching it to its consumer. The Ops lesson's chain of grants is, in Python, the lifetime ownership of the connection, the feature store, the model client; each stage receives the resources it needs as constructor arguments, and the principal (the run_pipeline driver) is the one that holds the original references.

The Fan-Out pattern does not map as cleanly to generators alone. A Fan-Out from a single generator to N consumers requires either tee-ing the upstream generator with itertools.tee or constructing N independent generators each driving its own variant of the downstream work. Python's concurrent.futures.ThreadPoolExecutor or asyncio.gather is the standard primitive for the orchestrator-dispatches-and-awaits shape.

The Specialist Team pattern is harder to model in pure Python primitives. The natural representation is a dictionary of named callable specialists with a coordinator function that routes by request type. Python's typing.Protocol classes let the coordinator hold specialists by structural interface rather than by concrete class. A future Dev lesson will treat the Specialist Team mapping in depth; this lesson stays on the Pipeline mapping where Python's idioms align most naturally.

The streaming-as-audit architectural payoff is what makes the Pipeline pattern attractive specifically in Python for ML inference. Buffering each intermediate stage's full output would defeat the streaming property; logging at each yield site preserves the streaming property and gains the audit-trail. This is the seam where the language-level mechanism and the architectural pattern align without compromise.

Paired Ops lesson → Archmagus-Stack/01-Earth-DevOps/Synthesis-Lessons/2026-05-18-multi-agent-orchestration-patterns-for-ml-training-workflows

§ VPrior-Lesson Reach

No prior Polyglot-Dev lessons exist yet; this is the first Dev lesson of the discipline. Backward-Synergy-Reach therefore points to the Python primitives this lesson presumed without re-deriving:

• The async/await model. Coroutines, event loops, awaitable objects, the cooperative concurrency model that asyncio provides. This lesson used async for and await without defining them. A future Python lesson on asyncio primitives will treat the model from the ground up.
• The typing module's Protocol class. Structural typing in Python; how a class can satisfy an interface without inheriting from it. Mentioned in §IV without elaboration. A future Python lesson on advanced typing will treat this in depth.
• itertools standard-library combinators. tee, chain, islice, takewhile, groupby. The functional toolkit Python provides for iterator manipulation. Touched-on in §IV; a future lesson treats them properly.

The forward-pointer is to a Python lesson on async-iteration composition: how to fan-out a single async generator to multiple async consumers, how to merge multiple async generators into a single stream, what the asyncio primitives as_completed, wait, and gather actually do under the hood. That lesson will close the Fan-Out gap this lesson left open.

§ VIClosing

Two things to hold from this lesson.

First, that the iterator protocol is not an exotic Python feature; it is the language-level mechanism for the Pipeline pattern. Every for loop in Python uses it. Every generator function is a Pipeline node in disguise. The engineer who has internalized this no longer thinks of generators as a memory-saving trick; they think of them as the syntactic form of sequential single-stage-to-next-stage data flow, which is the same shape the Ops lesson described at the architectural level.

Second, that the streaming-as-audit architectural payoff requires no extra machinery. A logger.info call before each yield is the audit. The Pipeline carries its own observability surface because the chain is already an event stream. Python makes this composable in a way few other languages do; the iterator protocol is the reason.

Write a five-stage generator chain. Add a logging wrapper to each stage. Watch the audit-trail emit as the stream flows. That is the lesson made concrete.