feat(evaluation) 3/15: indexing throughput aggregation

openrag/core/evaluation/__init__.py

@@ -1,9 +1,11 @@
 """Pure evaluation logic: test-set parsing, promptfoo config, metric math."""
 
 from core.evaluation.identity import sanitize_file_id
+from core.evaluation.metrics import indexing_metrics
 from core.evaluation.testset import parse_testset
 
 __all__ = [
+    "indexing_metrics",
     "parse_testset",
     "sanitize_file_id",
 ]

openrag/core/evaluation/metrics.py

@@ -0,0 +1,75 @@
+"""Metric computation for an evaluation run.
+
+Aggregates the per-file indexing timings the worker collected. Pure: the
+worker measures, this decides what the measurements mean.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Sequence
+from pathlib import Path
+
+from core.models.evaluation import (
+    FileIndexingSample,
+    IndexingMetrics,
+)
+
+_BYTES_PER_MB = 1024 * 1024
+
+
+def _percentile(values: Sequence[float], fraction: float) -> float:
+    """Nearest-rank percentile.
+
+    ``statistics.quantiles`` needs at least two points and interpolates;
+    nearest-rank keeps a single-file run meaningful and always returns a
+    duration that was actually observed.
+    """
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    # ceil, not round: the rank is ceil(fraction * n) by definition, and round()
+    # breaks ties to even, selecting the wrong element on an odd integer rank.
+    rank = math.ceil(fraction * len(ordered))
+    return float(ordered[min(max(rank, 1), len(ordered)) - 1])
+
+
+def indexing_metrics(samples: Sequence[FileIndexingSample], wall_seconds: float) -> IndexingMetrics:
+    """Aggregate per-file timings into throughput figures.
+
+    ``wall_seconds`` is the measured end-to-end duration of the indexing
+    phase, which is what throughput is derived from — summing per-file
+    durations would overstate speed whenever files are indexed concurrently.
+    """
+    succeeded = [s for s in samples if not s.failed]
+    durations = [s.duration_seconds for s in succeeded]
+    # Two different questions: how big the corpus is (every file the run
+    # attempted, matching ``files_total``) and how fast it moved (only what
+    # actually landed). A failed file has a size but contributed no throughput.
+    total_bytes = sum(s.size_bytes for s in samples)
+    indexed_bytes = sum(s.size_bytes for s in succeeded)
+
+    by_extension: dict[str, dict[str, float]] = {}
+    for sample in succeeded:
+        extension = (Path(sample.filename).suffix or "(none)").lower()
+        bucket = by_extension.setdefault(extension, {"files": 0.0, "seconds": 0.0})
+        bucket["files"] += 1
+        bucket["seconds"] += sample.duration_seconds
+    for bucket in by_extension.values():
+        bucket["mean_seconds"] = round(bucket["seconds"] / bucket["files"], 3)
+
+    return IndexingMetrics(
+        files_total=len(samples),
+        files_failed=sum(1 for s in samples if s.failed),
+        bytes_total=total_bytes,
+        wall_seconds=round(wall_seconds, 3),
+        files_per_minute=round(len(succeeded) / wall_seconds * 60, 2) if wall_seconds > 0 else 0.0,
+        megabytes_per_second=(round(indexed_bytes / _BYTES_PER_MB / wall_seconds, 3) if wall_seconds > 0 else 0.0),
+        p50_seconds=round(_percentile(durations, 0.50), 3),
+        p95_seconds=round(_percentile(durations, 0.95), 3),
+        by_extension=by_extension,
+        samples=list(samples),
+    )
+
+
+__all__ = ["indexing_metrics"]

tests/unit/core/evaluation/test_metrics.py

@@ -0,0 +1,73 @@
+"""Tests for indexing aggregation."""
+
+from __future__ import annotations
+
+from core.evaluation.metrics import indexing_metrics
+from core.models.evaluation import FileIndexingSample
+
+
+def _sample(name: str, seconds: float, size: int = 1024, failed: bool = False):
+    return FileIndexingSample(filename=name, size_bytes=size, duration_seconds=seconds, failed=failed)
+
+
+# ── indexing ─────────────────────────────────────────────────────────
+
+
+def test_throughput_uses_wall_clock_not_summed_durations():
+    """Files may be indexed concurrently, so summing per-file durations would
+    overstate throughput."""
+    metrics = indexing_metrics([_sample("a.pdf", 4.0), _sample("b.pdf", 4.0)], wall_seconds=4.0)
+    assert metrics.files_per_minute == 30.0
+
+
+def test_failed_files_are_counted_but_excluded_from_throughput():
+    """A failure is part of the corpus but contributed no throughput, so it
+    counts toward the totals and toward neither rate."""
+    metrics = indexing_metrics(
+        [_sample("a.pdf", 2.0, size=1024), _sample("b.pdf", 0.0, size=4096, failed=True)],
+        wall_seconds=2.0,
+    )
+    assert metrics.files_total == 2
+    assert metrics.files_failed == 1
+    assert metrics.files_per_minute == 30.0
+    # "Corpus size" in the UI — it must not shrink because a file failed.
+    assert metrics.bytes_total == 1024 + 4096
+    assert metrics.megabytes_per_second == round(1024 / (1024 * 1024) / 2.0, 3)
+
+
+def test_percentiles_on_a_single_file_return_that_file():
+    metrics = indexing_metrics([_sample("a.pdf", 3.0)], wall_seconds=3.0)
+    assert metrics.p50_seconds == 3.0
+    assert metrics.p95_seconds == 3.0
+
+
+def test_p50_of_an_even_sample_takes_the_lower_middle():
+    """Nearest-rank p50 is ceil(n/2); rounding half-to-even would report the
+    slower file for even n whose half is odd."""
+    metrics = indexing_metrics([_sample("a.pdf", 1.0), _sample("b.pdf", 10.0)], wall_seconds=11.0)
+    assert metrics.p50_seconds == 1.0
+    assert metrics.p95_seconds == 10.0
+
+
+def test_p50_ignores_files_that_failed_to_index():
+    metrics = indexing_metrics(
+        [_sample("a.pdf", 5.0), _sample("b.pdf", 0.0, failed=True)],
+        wall_seconds=5.0,
+    )
+    assert metrics.p50_seconds == 5.0
+
+
+def test_zero_wall_time_does_not_divide_by_zero():
+    metrics = indexing_metrics([_sample("a.pdf", 0.0)], wall_seconds=0.0)
+    assert metrics.files_per_minute == 0.0
+    assert metrics.megabytes_per_second == 0.0
+
+
+def test_breakdown_is_grouped_by_lowercased_extension():
+    metrics = indexing_metrics(
+        [_sample("a.PDF", 2.0), _sample("b.pdf", 4.0), _sample("c.txt", 1.0)],
+        wall_seconds=7.0,
+    )
+    assert metrics.by_extension[".pdf"]["files"] == 2
+    assert metrics.by_extension[".pdf"]["mean_seconds"] == 3.0
+    assert metrics.by_extension[".txt"]["files"] == 1