From e573a8aed0c372ca0592edf8de8ee6250014f9b2 Mon Sep 17 00:00:00 2001 From: EnjoyBacon7 <59032058+EnjoyBacon7@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:30:07 +0000 Subject: [PATCH] feat(evaluation): indexing throughput aggregation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `indexing_metrics` folds the per-file timings the worker collects into throughput figures: files/min, MB/s, p50/p95 and a per-extension breakdown. Two definitions worth stating, both covered by tests: - Throughput divides by the measured wall clock of the indexing phase, not by the sum of the per-file durations. Summing would overstate speed the moment files are indexed concurrently. - Percentiles are nearest-rank (`ceil`). `statistics.quantiles` needs two points and interpolates, so it cannot describe a one-file run; and `round` breaks ties to even, which selects the wrong observation whenever `fraction * n` lands on an odd integer. Files that failed to index are counted, then excluded from every rate and percentile — a failure is not a fast file. --- openrag/core/evaluation/__init__.py | 2 + openrag/core/evaluation/metrics.py | 75 ++++++++++++++++++++++ tests/unit/core/evaluation/test_metrics.py | 73 +++++++++++++++++++++ 3 files changed, 150 insertions(+) create mode 100644 openrag/core/evaluation/metrics.py create mode 100644 tests/unit/core/evaluation/test_metrics.py diff --git a/openrag/core/evaluation/__init__.py b/openrag/core/evaluation/__init__.py index ad6dc313d..660779b69 100644 --- a/openrag/core/evaluation/__init__.py +++ b/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", ] diff --git a/openrag/core/evaluation/metrics.py b/openrag/core/evaluation/metrics.py new file mode 100644 index 000000000..04bbc84a0 --- /dev/null +++ b/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"] diff --git a/tests/unit/core/evaluation/test_metrics.py b/tests/unit/core/evaluation/test_metrics.py new file mode 100644 index 000000000..3bfa9bc87 --- /dev/null +++ b/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