ML Foundations
Sequential tee consumers can buffer a streamed dataset
Splitting a record stream with itertools.tee does not guarantee bounded storage. Check consumer lag and compute compatible metrics in one pass.
Evidence and testing
Review record: Tensor Almanac automated evidence gate (no independent human review). Example versions, recorded outputs and limitations appear below.
The slow consumer determines what must remain available
Splitting a streamed evaluation dataset into loss and accuracy iterators can quietly undermine the intended streaming behavior. Python itertools.tee provides separate iterators over one input, but it may require substantial auxiliary storage. The documentation specifically calls out the case where one consumer uses most or all records before another starts. [1]
If the loss pass finishes while the accuracy iterator has not advanced, the second consumer still needs the records already produced. Laziness at construction does not prevent this backlog. The example deliberately schedules the consumers sequentially to expose that condition. This is a storage implication of the documented tee behavior, not a measured byte count.
References: [1]
Count source yields across both passes
Run python examples/tee_stream_metrics.py. The source generates six small records, each containing a loss and a correctness flag. A counter increments at every source yield. Constructing the tee branches produces nothing. Completing the loss branch produces all six records before the accuracy branch begins.
The accuracy branch then obtains four correct results without increasing the source counter beyond six. Both consumers therefore receive the records from a single production pass. The second pass is replaying retained data rather than rerunning the generator. Assertions check the counters and aggregate results; the fixture does not inspect tee internals or measure process memory.
The loss reference uses math.fsum over the first branch and divides by the known six-record fixture size. Python documents fsum as an accurate floating-point summation over an iterable. It can consume an iterator without requiring the caller to construct a loss list. [2]
Executed with Python 3.12.14 using only standard-library itertools and math. Exact stdout and source-byte SHA-256 are preserved in private evidence.
"""Sequential tee consumers versus one-pass aggregate metrics."""
from itertools import tee
from math import fsum, isclose
def records(state):
# Generate one record at a time; the source counter records actual yields.
for i in range(6):
state["produced"] += 1
yield (i + 1) / 5, i not in (1, 4)
def one_pass_metrics(stream):
count = 0
loss_total = 0.0
correct_total = 0
for loss, correct in stream:
count += 1
loss_total += loss
correct_total += correct
if count == 0:
raise ValueError("at least one record required")
return loss_total / count, correct_total / count
def main():
state = {"produced": 0}
loss_stream, accuracy_stream = tee(records(state), 2)
assert state["produced"] == 0
# Fully consume the first branch before starting the second.
loss_mean = fsum(loss for loss, correct in loss_stream) / 6
after_first = state["produced"]
assert after_first == 6
correct_count = sum(correct for loss, correct in accuracy_stream)
after_second = state["produced"]
assert after_second == 6 and correct_count == 4
replay_metrics = (loss_mean, correct_count / 6)
single_state = {"produced": 0}
single_metrics = one_pass_metrics(records(single_state))
assert single_state["produced"] == 6
assert all(isclose(a, b) for a, b in zip(replay_metrics, single_metrics, strict=True))
assert isclose(single_metrics[0], 0.7)
assert isclose(single_metrics[1], 2 / 3)
empty_rejected = False
try:
one_pass_metrics(iter(()))
except ValueError:
empty_rejected = True
assert empty_rejected
print("source yields before consuming tee branches:", 0)
print("source yields after first / second full pass:", after_first, after_second)
print("tee mean loss / accuracy:", *(f"{v:.6f}" for v in replay_metrics))
print("one-pass mean loss / accuracy:", *(f"{v:.6f}" for v in single_metrics))
print("one-pass source yields:", single_state["produced"])
print("empty stream rejected:", empty_rejected)
if __name__ == "__main__":
main()
Recorded output
source yields before consuming tee branches: 0 source yields after first / second full pass: 6 6 tee mean loss / accuracy: 0.700000 0.666667 one-pass mean loss / accuracy: 0.700000 0.666667 one-pass source yields: 6 empty stream rejected: True
Reduce compatible metrics together
The alternative updates count, loss total and correct count inside one loop. It yields mean loss 0.7 and accuracy two-thirds, matching the tee calculation within floating-point tolerance, while generating six records once. The reducer keeps aggregate scalars instead of records awaiting another metric pass. Its empty-stream policy is an explicit ValueError, also exercised by the example.
This reducer uses ordinary floating-point addition; agreement on six moderate losses is not a general accuracy equivalence with fsum. Large or ill-conditioned sums need a suitable numerical accumulation strategy. The example demonstrates consumer scheduling and aggregate state, not throughput or a peak-memory benchmark.
When a metric needs the complete record history, these three scalars are insufficient. Choose explicit storage, a restartable data source, or a deliberately bounded processing design for that requirement. Tee remains useful when its storage behavior fits the consumers; splitting a stream alone does not establish a memory bound. The fixture is single-threaded and makes no concurrency guarantee.