Pipeline Stage Review and Reuse

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Iterators Laziness Streaming Dataflow"]
  page["Pipeline Stage Review and Reuse"]
  capstone["Capstone evidence"]

  family --> program --> section --> page
  page -.applies in.-> capstone

flowchart LR
  orient["Orient on the page map"] --> read["Read the main claim and examples"]
  read --> inspect["Inspect the related code, proof, or capstone surface"]
  inspect --> verify["Run or review the verification path"]
  verify --> apply["Apply the idea back to the module and capstone"]

Read the first diagram as a placement map: this page is one concept inside its parent module, not a detached essay, and the capstone is the pressure test for whether the idea holds. Read the second diagram as the working rhythm for the page: name the problem, study the example, identify the boundary, then carry one review question forward.

This lesson closes the reusable-stage hotspot. The main lesson should teach you how to build stage factories. This companion page explains how to review those factories and how to decide whether the extra indirection is actually helping.

Review route

Ask these questions of any reusable stage:

• does each factory call return a fresh iterator or transform?
• is the captured configuration immutable and explicit?
• can the stage still be explained as a simple input-output contract?
• did the abstraction remove duplication or just move it?

Review contracts, not factory syntax

Reusable stages earn their keep when these claims hold:

• two runs with equal input and equal config return equal output
• calling the same factory twice yields independent executions
• a fenced stage returns the same prefix as the unfenced baseline
• the generic stage and the hand-written domain-specific stage stay equivalent

from funcpipe_rag import as_source, make_chain

merged = make_chain(as_source((1, 2)), as_source((3,)))

assert list(merged()) == [1, 2, 3]
assert list(merged()) == [1, 2, 3]

Source[T] means Callable[[], Iterator[T]]. Calling a source starts a traversal. Transform[A, B] means Callable[[Iterable[A]], Iterator[B]]. Calling a transform wraps a supplied traversal.

Question	Source[T]	Transform[A, B]
Who owns input creation?	source	caller
What should a call create?	fresh input iterator	fresh output traversal
FuncPipe example	as_source(docs)	fence_k(100)
Common bug	captured exhausted iterator	retained mutable per-run state

Determinism requires replayable input. Calling a transform twice with the same already-consumed generator is not a valid determinism test. Use a tuple, list, or fresh source call for each run.

Audit make_gen_rag_fn

The real Module 03 signature captures chunk_size, max_chunks, cleaning config, and keep rules, then returns a function from an Iterable[RawDoc] to an iterator of final unique Chunk values. It earns the abstraction because it:

• wires a stable RAG configuration once;
• leaves documents as the explicit varying input;
• creates stream_rag_chunks inside every run;
• includes deterministic embedding and encounter-order structural deduplication;
• exposes the returned iterator rather than silently collecting it.

It does not promise restartability for an input generator supplied by the caller. The factory owns output traversal state, not the caller's input.

max_chunks belongs after deduplication in this factory because it describes final unique outputs. It may therefore pull more raw chunks than it emits. Use gen_bounded_chunks directly when the contract is instead “perform at most this many pre-embedding chunk operations.”

Audit stream_rag_sources

stream_rag_sources is an application composition function, not another generic fan-in factory. Its inputs make ownership visible:

stream_rag_sources(
    sources: Sequence[Source[RawDoc]],
    config: RagConfig,
    deps: RagCoreDeps,
    *,
    schedule: SourceSchedule = "sequential",
    max_chunks: int | None = None,
) -> Iterator[Chunk]

• each Source[RawDoc] owns creation of one fresh document iterator;
• the caller owns the immutable sequence of source factories;
• SourceSchedule owns the cross-source encounter-order policy;
• stream_rag_sources owns composition into the complete RAG chain;
• the returned iterator owns one execution and cannot be restarted.

The function earns a RAG-specific boundary because it rules out underspecified generic choices. It offers sequential and round-robin scheduling, whose preconditions it can state locally. It does not expose make_merge, because sorted fan-in requires every document source to satisfy the same key-order precondition. That proof belongs at the adapters creating those sources.

Review the two schedules by values and demand:

Question	Sequential	Round robin
Cross-source output	exhaust each source in order	one item from each active source per cycle
First demand opens	first source only	every source iterator
Per-source order	preserved	preserved
Long first source	delays later sources	later active sources get turns
Blocking next()	blocks the route	still blocks the route

Avoid a boolean such as fair=True. “Fair” would conceal whether the promise is synchronous turn-taking, non-blocking I/O, weighted priority, or async scheduling. The literal policy names the only behavior Module 03 implements.

Test values and freshness separately

def test_factory_value_contract():
    run = make_gen_rag_fn(chunk_size=50, max_chunks=3)
    assert list(run(replayable_docs)) == list(run(replayable_docs))


def test_factory_starts_no_work():
    run = make_gen_rag_fn(chunk_size=50, max_chunks=3)
    stream = run(counted_documents())
    assert requested_documents == 0
    next(stream)
    assert requested_documents == 1

The first test proves deterministic values for replayable input. The second proves execution timing. Neither alone proves that the iterator completes the RAG chain. Also assert that returned values have embeddings and compare the unfenced, canonically ordered case with full_rag_api_docs.

For multiple sources, add a third category: factory-opening timing. A tracked source should record once when called and again immediately before each document yield. That distinguishes round-robin creating every iterator from actually pulling a document from every iterator.

When reuse is worth it

Keep the factory form when:

• the same transformation logic appears in more than one route
• the configuration changes while the stage logic stays the same
• tests need to compare multiple variants cheaply
• the abstraction clarifies the pipeline taxonomy of source, transform, and sink

Do not force it when:

• the code is genuinely one local step
• the factory names are more abstract than the actual work
• the captured configuration is so large nobody can explain it quickly

Capstone check

Before moving on:

1. inspect streaming/types.py, streaming/fanin.py, api/core.py, and api/config.py in the Module 03 reference state;
2. classify as_source, fence_k, stream_rag_sources, and make_gen_rag_fn by ownership;
3. run the multi-source proof:

make PROGRAM=python-programming/python-functional-programming \
  capstone-source-scheduling-proof

1. run the complete streaming proof:

make PROGRAM=python-programming/python-functional-programming \
  capstone-streaming-rag-proof

1. distinguish the raw-chunk fence from the unique-output fence;
2. reject any abstraction whose demand and retained-state behavior cannot be stated more simply than its implementation.

Reflection

• Which repeated stage in your own codebase should become a factory?
• Which current factory should collapse back into a direct function?
• Which config object is being captured only because the stage boundary is still unclear?

Continue with: Fan-In and Fan-Out