Introducing Laziness

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Data First Apis Expression Style"]
  page["Introducing Laziness"]
  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"]

This lesson becomes important the moment a pipeline is conceptually simple but operationally wasteful. If the code keeps building full intermediate lists just to throw most of them away, laziness is not an optimization trick anymore. It is the clearer design.

Start With the Waste Pattern

Generators are often introduced as syntax first and purpose second. The purpose here is simpler: stop doing all the work up front when the consumer only needs a stream.

• If each stage returns a new list, ask whether the next stage really needs the whole collection.
• If memory usage grows with every intermediate transformation, the pipeline boundary is probably in the wrong place.
• If the code only needs a prefix or one pass, eager materialization is hiding unnecessary work.

Keep This Question In View

Core question:
How do you replace eager, memory-hungry list comprehensions with lazy generators—so pipelines stay efficient, composable, and only compute what’s needed?

This lesson introduces laziness in the way you need to apply it:

• treat intermediate results as on-demand streams instead of permanent collections
• keep generator-based stages composable with the configurators and expressions introduced earlier
• choose one place where materialization is allowed instead of letting it leak everywhere

The running project keeps the lesson honest: laziness is useful when it preserves a readable pipeline while reducing waste, not when it turns every example into a generator puzzle.

Use this when you have expression-oriented pipelines but still materialize large lists with [...] comprehensions, risking unnecessary memory use and accidental eager work. Outcome:
1. Spot eager materialization in code and explain why it wastes memory.
2. Refactor an eager list comprehension to a lazy generator.
3. Write a Hypothesis property providing strong evidence of equivalence on finite data, including a shrinking example.

---

Runnability Note (Module 01 Snapshot vs Module 02 End-State)

Some “before” snippets in this core are hypothetical pre-refactor examples used for contrast. They are labeled accordingly and are not meant to exactly match a real snapshot. We refactor these shapes into the real Module 02 API as the module progresses.

For a real, runnable Module 01 codebase, refresh the generated history route first:

• make PROGRAM=python-programming/python-functional-programming history-refresh
• Module 01 path: capstone/_history/worktrees/module-01/
• Import path for Module 01: capstone/_history/worktrees/module-01/src/

1. Conceptual Foundation

1.1 Laziness with Generators in One Precise Sentence

Laziness with generators defers computation until values are needed, using yield to produce sequences on-demand—avoiding memory allocation for large or infinite data.

1.2 The One-Sentence Rule

Default to generators for large or unbounded data; materialize only at well-defined edges.

1.3 Why This Matters Now

Expression-oriented code made pipeline steps easier to see, but it did not stop those steps from eagerly building whole collections. This page addresses that gap. A lazy stage preserves the same visible dataflow while postponing work until the next item is actually requested. That is the key distinction to understand before you judge when laziness helps and when it only adds complexity.

1.4 Laziness as Values in 5 Lines

The next snippet matters because it turns a chunking strategy into a reusable stream-producing value rather than a list-building helper.

from collections.abc import Callable, Generator
from funcpipe_rag import CleanDoc, ChunkWithoutEmbedding, RagEnv
from itertools import islice


def chunk_stream(doc: CleanDoc, env: RagEnv) -> Generator[ChunkWithoutEmbedding, None, None]:
    text = doc.abstract
    for i in range(0, len(text), env.chunk_size):
        chunk_text = text[i:i + env.chunk_size]
        if not chunk_text:
            break
        yield ChunkWithoutEmbedding(doc.doc_id, chunk_text, i, i + len(chunk_text))


StreamFactory = Callable[[CleanDoc, RagEnv], Generator[ChunkWithoutEmbedding, None, None]]

streams: dict[str, StreamFactory] = {
    "chunks": chunk_stream,
    # Add more streams
}


def consume_stream(key: str, doc: CleanDoc, env: RagEnv, n: int) -> list[ChunkWithoutEmbedding]:
    it = streams[key](doc, env)
    return list(islice(it, n))

Because generators are lazy (compute on next), we can safely store and compose them with Core 1 partials—just like data. Note: generators are one-shot; use factories for reuse.

---

2. Mental Model: Eager Lists vs Lazy Generators

2.1 One Picture

Eager Lists (Memory-Bound)              Lazy Generators (Streaming)
+-----------------------+               +------------------------------+
| huge = [x for x in N] |               |   huge = (x for x in N)      |
| # OOM on large N      |               |   # O(1) memory              |
| print(huge[0])        |               |   print(next(huge))          |
+-----------------------+               +------------------------------+
   ↑ Allocates All Now                     ↑ Computes On-Demand

2.2 Contract Table

Aspect	Eager Lists	Lazy Generators
Memory	O(n) allocation	O(1) constant
Computation	Upfront all	On-demand per item
Infinite Data	Impossible	Safe with islice
Composability	Lists chain eagerly	Generators chain lazily
Testing	Finite only	Finite prefixes via islice

Note on Eager Choice: Rarely, for small profiled data (e.g., cache reuse), materialize behind a lazy API.

---

3. Running Project: FuncPipe RAG Builder

Our running project (from m02-rag.md) is extending the pure RAG pipeline from Module 1 with laziness.
- Dataset: 10k arXiv CS abstracts (arxiv_cs_abstracts_10k.csv).
- Goal: Make the internal pipeline lazy by replacing eager lists with generators; we still materialize at the edges for now (e.g., for deduplication). Module 3 will fully generalize lazy streaming.
- Start: Hypothetical pre-refactor eager version (core3_start.py, illustration only).
- End (this core): Lazy core, preserving equivalence to Module 1.

3.1 Types (Canonical, Used Throughout)

From capstone/src/funcpipe_rag/rag_types.py and capstone/src/funcpipe_rag/api/types.py (as in Core 1/2).

3.2 Eager Start (Anti-Pattern)

This is a hypothetical pre-refactor example used for contrast. It is intentionally not intended to be run as-is in the end-of-Module-02 checkout.

# core3_start.py (hypothetical pre-refactor; illustration only)
from funcpipe_rag import RawDoc, CleanDoc, ChunkWithoutEmbedding, Chunk, RagEnv
from funcpipe_rag import DocRule, Observations, RagTaps
from funcpipe_rag import any_doc
from funcpipe_rag import clean_doc, embed_chunk, structural_dedup_chunks
from typing import Callable


def eager_full_rag_api(docs: list[RawDoc], env: RagEnv, cleaner: Callable[[RawDoc], CleanDoc], *,
                       keep: DocRule | None = None, taps: RagTaps | None = None) -> tuple[list[Chunk], Observations]:
    rule = keep if keep is not None else any_doc
    kept_docs = [d for d in docs if rule(d)]  # Materializes full list
    if taps and taps.docs:
        taps.docs(tuple(kept_docs))
    cleaned = [cleaner(d) for d in kept_docs]  # Another full list
    if taps and taps.cleaned:
        taps.cleaned(tuple(cleaned))
    chunk_we = [c for cd in cleaned for c in gen_chunk_doc(cd, env)]  # Materializes huge chunks
    embedded = [embed_chunk(c) for c in chunk_we]  # Full embed list
    chunks = structural_dedup_chunks(embedded)
    if taps and taps.chunks:
        taps.chunks(tuple(chunks))
    obs = Observations(
        total_docs=len(docs),
        total_chunks=len(chunks),
        kept_docs=len(kept_docs),
        cleaned_docs=len(cleaned),
        sample_doc_ids=tuple(d.doc_id for d in kept_docs[:5]),
        sample_chunk_starts=tuple(c.start for c in chunks[:5]),
    )
    return chunks, obs


# Usage: Risks OOM on large docs
docs: list[RawDoc] = [RawDoc("cs-123", "Title", "Abstract text...", "cs.AI")]
chunks1, obs1 = eager_full_rag_api(docs, RagEnv(512), clean_doc)
chunks2, obs2 = eager_full_rag_api(docs, RagEnv(512), clean_doc)
assert chunks1 == chunks2

Smells: Eager lists (kept_docs, cleaned, chunk_we), upfront allocation.
Problem: Materializes intermediates; OOM on big data.

---

4. Refactor to Lazy: Generators and Yield

4.1 Lazy Core

First, the basic transformation from eager list to lazy generator:

from itertools import islice

# Eager
squares = [x**2 for x in range(1000000)]  # Allocates full list in memory

# Lazy
squares_gen = (x**2 for x in range(1000000))  # O(1) memory
first_few = list(islice(squares_gen, 10))  # Materialize only what's needed

Now apply to RAG: Use generators; defer computation. Define gen_chunk_doc as a generator.

from collections.abc import Generator, Iterable, Iterator, Callable
from funcpipe_rag import RawDoc, CleanDoc, ChunkWithoutEmbedding, Chunk, RagEnv
from funcpipe_rag import DocRule
from funcpipe_rag import any_doc
from funcpipe_rag import clean_doc, embed_chunk, structural_dedup_chunks


def gen_chunk_doc(cd: CleanDoc, env: RagEnv) -> Generator[ChunkWithoutEmbedding, None, None]:
    text = cd.abstract
    for start in range(0, len(text), env.chunk_size):
        chunk_text = text[start: start + env.chunk_size]
        if chunk_text:
            yield ChunkWithoutEmbedding(cd.doc_id, chunk_text, start, start + len(chunk_text))


def iter_rag(
        docs: Iterable[RawDoc],
        env: RagEnv,
        cleaner: Callable[[RawDoc], CleanDoc],
        *,
        keep: DocRule | None = None,
) -> Iterator[Chunk]:
    rule = keep if keep is not None else any_doc
    kept_docs_gen = (d for d in docs if rule(d))  # Lazy filter
    cleaned_gen = (cleaner(d) for d in kept_docs_gen)  # Lazy map
    chunk_we_gen = (c for cd in cleaned_gen for c in gen_chunk_doc(cd, env))  # Lazy flatMap
    embedded_gen = (embed_chunk(c) for c in chunk_we_gen)  # Lazy map
    yield from embedded_gen  # Stream undeduped embedded chunks


# Lazy pipeline internally; materialize at edges (e.g., for dedup)
# Boundary provides a finite, re-iterable input (e.g., a list of RawDoc).
docs: list[RawDoc] = [RawDoc("cs-123", "Title", "Abstract text...", "cs.AI")]
stream = iter_rag(docs, RagEnv(512), clean_doc)
# For deduplication (requires full view), materialize:
chunks1 = structural_dedup_chunks(list(stream))
# Here docs is the boundary: we assume a finite dataset. Module 3 covers streaming from disk/network lazily as well.
chunks2 = structural_dedup_chunks(list(iter_rag(docs, RagEnv(512), clean_doc)))
assert chunks1 == chunks2

Wins: Lazy generators chain with O(1) memory until materialization; matches Module 1/Core 1/2 semantics.
Note: Deduplication requires the full list (global view), so we materialize there; upstream remains lazy. Consume at edges; compose with Module 3 for advanced streaming. Module 3 will generalise iter_rag into reusable iterator stages and bring in itertools (chain, groupby, islice, etc.) — here we only care about the basic [...] → (...) refactor.

---

5. Equational Reasoning: Substitution Exercise

Hand Exercise: Replace expressions in iter_rag.
1. Inline kept_docs_gen = (d for d in docs if rule(d)) → lazy filter.
2. Substitute into cleaned_gen → lazy map.
3. Result: Entire stream computes on-demand until edge materialization.
Bug Hunt: In eager version, substitution allocates eagerly.

---

6. Property-Based Testing: Providing Strong Evidence of Equivalence (Advanced, Optional)

Use Hypothesis to provide strong evidence that the refactor preserved behavior to Module 1.

6.1 Custom Strategy (RAG Domain)

From capstone/tests/conftest.py (as in Module 1).

6.2 Equivalence Property

# capstone/tests/test_rag_api.py
from hypothesis import given
import hypothesis.strategies as st
from funcpipe_rag import (
    RagEnv,
    clean_doc,
    embed_chunk,
    iter_chunk_doc,
    iter_rag,
    structural_dedup_chunks,
)
from tests.conftest import doc_list_strategy, env_strategy

def baseline_full_rag(docs, env):
    embedded = [embed_chunk(c) for d in docs for c in iter_chunk_doc(clean_doc(d), env)]
    return structural_dedup_chunks(embedded)

@given(docs=doc_list_strategy(), env=env_strategy())
def test_m02c03_iter_rag_equivalence(docs, env):
    # Lazy: materializes only for dedup (necessary for global view)
    lazy_stream = iter_rag(docs, env, clean_doc)
    deduped_lazy = structural_dedup_chunks(list(lazy_stream))
    # Equivalence to Module 1 (full pipeline)
    assert deduped_lazy == baseline_full_rag(docs, env)

Note: On small finite data (per strategy), full materialization in test is fine; proves semantic preservation.

6.3 Shrinking Demo: Catching a Bug

Bad refactor (reusing consumed iterator):

def bad_iter_rag(docs: Iterable[RawDoc], env: RagEnv, cleaner: Callable[[RawDoc], CleanDoc], *, keep: DocRule | None = None) -> Iterator[Chunk]:
    rule = keep if keep is not None else any_doc
    kept_docs_gen = (d for d in docs if rule(d))  # Lazy
    list(kept_docs_gen)  # Accidentally consumes (e.g., for debug print/logging)
    cleaned_gen = (cleaner(d) for d in kept_docs_gen)  # Now exhausted!
    chunk_we_gen = (c for cd in cleaned_gen for c in gen_chunk_doc(cd, env))  # Empty
    embedded_gen = (embed_chunk(c) for c in chunk_we_gen)  # Empty
    yield from embedded_gen  # Empty stream

Property (swapped to bad_iter_rag):

@given(docs=doc_list_strategy(), env=env_strategy())
def test_bad_rag(docs, env):
    module1_chunks = baseline_full_rag(docs, env)
    bad_stream = bad_iter_rag(docs, env, clean_doc)
    deduped_bad = structural_dedup_chunks(list(bad_stream))
    assert deduped_bad == module1_chunks  # Fails: empty vs non-empty

Hypothesis failure trace (run to verify; example):

Falsifying example: test_bad_rag(
    docs=[RawDoc(doc_id='a', title='', abstract='a', categories='')], 
    env=RagEnv(chunk_size=128),
)
AssertionError

• Shrinks to minimal non-empty docs; catches exhaustion bug (empty output despite valid input).

---

7. When Laziness Isn't Worth It

Rarely, for small/hot paths where full materialization is cheaper, use lists behind a lazy API.

---

8. Pre-Core Quiz

1. Eager [...] on huge → violates? → No eager materialization
2. Double list build → violates? → One-pass only

3. sorted(infinite) → what’s the correct approach?

   Don’t sort the whole infinite stream. Work on finite windows or partial orderings: - sorted(islice(infinite_stream, n)) to sort a finite prefix. - heapq.nsmallest / nlargest for top-k without full sort. - heapq.merge only to merge multiple already-sorted (possibly infinite) streams. 4. Mid-pipeline list() → fix with? → Consume at edge
   5. Tool to prove lazy ≡ eager (finite)? → Hypothesis equivalence

9. Post-Core Reflection & Exercise

Reflect: In your code, find one eager list on big/streaming data.
Apply the recipe: 1. Replace with generator expression / yield. 2. Chain lazy operations. 3. Materialize only at edge (e.g., for global ops like dedup). 4. Prove equivalence with Hypothesis. 5. Compose with Core 1/2 patterns.

Did memory usage drop?
Did the pipeline handle larger data?

Project Exercise: Apply to RAG; run properties on sample data.

Continue with: FP-Friendly APIs

Verify all patterns with Hypothesis—examples provided show how to detect impurities like globals or non-determinism.

Further Reading: For more on generators in Python, see 'Fluent Python' by Luciano Ramalho. Explore itertools for advanced lazy tools once comfortable.