Python 10：惰性管道在哪里积累成本

词元改成生成器后，还要过滤停用词并分批消费。把数据分成两块，第一块里正好有两个词；取出第一批时如果第二块也被读取，管道就有了不必要的提前工作。只看最终词频相同，无法发现上游是否被额外消费。

09 用读取块记录验证过第一次 next 的范围；本章继续在 tasklab 上组合过滤与批处理，同时保持未过滤任务的词频业务合同。

先知道过滤针对谁

word_batches(chunks, skip=(), size=64) 将 iter_words(chunks) 产生的词元交给生成器表达式筛选，再用 Python 3.12 标准库 itertools.batched 组成固定容量的元组批次。核心路径是：

1
2
3

skipped = frozenset(skip)
words = (word for word in iter_words(chunks) if word not in skipped)
return batched(words, size)

过滤条件作用于单个词元，不是整份任务结果；skip={"skip"} 时得到 "one" 两次、"two" 一次。正常的 run 不传过滤词，所以 "one two one skip" 的业务结果仍为 {"one": 2, "two": 1, "skip": 1}。若把筛选偷偷用于所有任务，就破坏了从 00 开始的词频合同。

初始化批处理入口时先检查 size >= 1；size=0 直接抛 ValueError，上游块读取计数仍为零。这条检查防止把无效参数错误拖到部分输入已被读取之后。停用词容器在创建批处理对象时转为 frozenset，所以调用后继续改外部停用词列表也不会改变正在执行的这一轮过滤规则。

用读取次数检验“按需”

给 word_batches 两段合成输入 "one two " 和 "one skip "，指定 skip={"skip"} 与批次大小 2。第一次 next(batches) 返回 ("one", "two")，此时记录的已读取块只有 "one two "。再消费余下批次才读到第二块，最终读入记录里每块各出现一次。实验 JSON 将元组编码成数组，因此原始标准输出中的首批显示为 ["one", "two"]；这不是应用把批次类型改为列表。

管道惰性只保证“需求到来前不做尚不必做的上游读取”，不保证每次操作的成本为零。frozenset(skip) 先分配过滤表，batched 内部需要凑足一批，iter_words 仍暂存尚未遇到空白的一个词；若上游有副作用，重复建立一条新的管道并重新遍历也会重复触发副作用。本章只验证同一条管道的一次消费没有重复读取输入块，不能推广为所有计算都只执行一次。

在 examples/python-language/ 运行：

1 2	`PYTHONPATH=src /usr/bin/python3.12 experiments/10_pipeline.py PYTHONPATH=src /usr/bin/python3.12 -m unittest discover -s tests -v`

演示与累计回归均以 0 退出，共执行 32 条测试。新增断言覆盖每块恰好读取一次、未过滤任务与原结果一致、无效批次大小未触发上游读取。examples/python-language/evidence/10/20261007-pipeline/ 保留原始输出、命令、基线环境与断言。没有计时或内存样本，因而这里不宣称“批处理比单词逐个遍历更快”；这样的比较要留给 36。