Add performance benchmarks for the non-Analyzer pipeline (MASTER_SPEC §26.6, NFR-004) plus tests. - scripts/benchmark.py: measures Knowledge Base load time, single-scene resolution time, and full pipeline time (median over iterations) and records a baseline to docs/benchmarks.md. - docs/benchmarks.md: recorded baseline numbers (indicative, machine-dependent). - tests/test_performance.py: a benchmark smoke test (functions run, all metrics positive, a generous non-flaky bound), a check that the recorded baseline exists, and two tests confirming the Knowledge Base is loaded once and reused (the loader caches; the pipeline never reloads per compilation). Co-authored-by: kunail0804 <kunail0804@gmail.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
kunail0804
Claude Opus 4.8
parent
584cffb1c5
commit
7ebe8978ca
@@ -0,0 +1,10 @@
|
||||
# Performance baselines
|
||||
|
||||
Median wall-clock times for the non-Analyzer pipeline (Knowledge Base: 188 entries; scene: `complex_multi_character`).
|
||||
Numbers are machine-dependent and indicative; regenerate with `python scripts/benchmark.py`.
|
||||
|
||||
| Metric | Median |
|
||||
|---|---|
|
||||
| Knowledge Base load | 11.26 ms |
|
||||
| Single-scene resolution | 57.6 µs |
|
||||
| Full pipeline (validate → build) | 742.6 µs |
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Benchmark the non-Analyzer compiler pipeline (issue #29, MASTER_SPEC §26.6, NFR-004).
|
||||
|
||||
Measures Knowledge Base load time, single-scene resolution time, and full
|
||||
non-Analyzer pipeline time (median over several iterations). Prints a table and
|
||||
records a baseline to ``docs/benchmarks.md`` for future comparison.
|
||||
|
||||
Usage:
|
||||
python scripts/benchmark.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import statistics
|
||||
import sys
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from compiler.builder.prompt_builder import build_prompts # noqa: E402
|
||||
from compiler.common.config import Config # noqa: E402
|
||||
from compiler.common.knowledge_base import KnowledgeBase, load_knowledge_base # noqa: E402
|
||||
from compiler.resolver.illustrious_resolver import resolve_scene # noqa: E402
|
||||
from compiler.splitter.category_splitter import split_into_categories # noqa: E402
|
||||
from compiler.validator.scene_validator import validate_scene # noqa: E402
|
||||
from tests.regression.golden_scenes import KB_DIR, SCENES # noqa: E402
|
||||
|
||||
_BENCHMARK_SCENE = "complex_multi_character"
|
||||
_BASELINE_FILE = Path(__file__).resolve().parent.parent / "docs" / "benchmarks.md"
|
||||
|
||||
|
||||
def _median_seconds(operation: Callable[[], object], iterations: int) -> float:
|
||||
"""Return the median wall-clock time of ``operation`` over ``iterations`` runs."""
|
||||
samples = []
|
||||
for _ in range(iterations):
|
||||
start = time.perf_counter()
|
||||
operation()
|
||||
samples.append(time.perf_counter() - start)
|
||||
return statistics.median(samples)
|
||||
|
||||
|
||||
def measure_kb_load(iterations: int = 20) -> float:
|
||||
return _median_seconds(lambda: load_knowledge_base(KB_DIR), iterations)
|
||||
|
||||
|
||||
def measure_resolution(kb: KnowledgeBase, scene: dict, iterations: int = 200) -> float:
|
||||
config = Config()
|
||||
validated = validate_scene(scene, config).data
|
||||
return _median_seconds(lambda: resolve_scene(validated, kb, config), iterations)
|
||||
|
||||
|
||||
def measure_full_pipeline(kb: KnowledgeBase, scene: dict, iterations: int = 200) -> float:
|
||||
config = Config()
|
||||
|
||||
def run() -> None:
|
||||
validated = validate_scene(scene, config)
|
||||
resolved = resolve_scene(validated.data, kb, config)
|
||||
categorized = split_into_categories(resolved.data)
|
||||
build_prompts(categorized.data, config)
|
||||
|
||||
return _median_seconds(run, iterations)
|
||||
|
||||
|
||||
def run_benchmarks() -> dict[str, float]:
|
||||
"""Run all benchmarks and return their median seconds."""
|
||||
kb = load_knowledge_base(KB_DIR)
|
||||
scene = SCENES[_BENCHMARK_SCENE]
|
||||
return {
|
||||
"knowledge_base_load": measure_kb_load(),
|
||||
"single_scene_resolution": measure_resolution(kb, scene),
|
||||
"full_pipeline": measure_full_pipeline(kb, scene),
|
||||
}
|
||||
|
||||
|
||||
def _format_baseline(results: dict[str, float], entry_count: int) -> str:
|
||||
lines = [
|
||||
"# Performance baselines",
|
||||
"",
|
||||
"Median wall-clock times for the non-Analyzer pipeline "
|
||||
f"(Knowledge Base: {entry_count} entries; scene: `{_BENCHMARK_SCENE}`).",
|
||||
"Numbers are machine-dependent and indicative; regenerate with "
|
||||
"`python scripts/benchmark.py`.",
|
||||
"",
|
||||
"| Metric | Median |",
|
||||
"|---|---|",
|
||||
f"| Knowledge Base load | {results['knowledge_base_load'] * 1e3:.2f} ms |",
|
||||
f"| Single-scene resolution | {results['single_scene_resolution'] * 1e6:.1f} µs |",
|
||||
f"| Full pipeline (validate → build) | {results['full_pipeline'] * 1e6:.1f} µs |",
|
||||
"",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
results = run_benchmarks()
|
||||
entry_count = len(load_knowledge_base(KB_DIR))
|
||||
baseline = _format_baseline(results, entry_count)
|
||||
_BASELINE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
_BASELINE_FILE.write_text(baseline, encoding="utf-8")
|
||||
sys.stdout.write(baseline)
|
||||
sys.stdout.write(f"\nWrote baseline to {_BASELINE_FILE}\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Performance tests and benchmark smoke test (issue #29, §26.6, NFR-004)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import compiler.common.knowledge_base as kb_module
|
||||
from scripts.benchmark import (
|
||||
measure_full_pipeline,
|
||||
measure_kb_load,
|
||||
measure_resolution,
|
||||
run_benchmarks,
|
||||
)
|
||||
from tests.regression.golden_scenes import (
|
||||
SCENES,
|
||||
compile_prompt_outputs,
|
||||
load_reference_knowledge_base,
|
||||
)
|
||||
|
||||
BASELINE_FILE = Path(__file__).resolve().parent.parent / "docs" / "benchmarks.md"
|
||||
|
||||
|
||||
def test_benchmark_functions_return_positive_durations() -> None:
|
||||
kb = load_reference_knowledge_base()
|
||||
scene = SCENES["complex_multi_character"]
|
||||
assert measure_kb_load(iterations=2) > 0
|
||||
assert measure_resolution(kb, scene, iterations=5) > 0
|
||||
assert measure_full_pipeline(kb, scene, iterations=5) > 0
|
||||
|
||||
|
||||
def test_run_benchmarks_reports_all_metrics() -> None:
|
||||
results = run_benchmarks()
|
||||
assert set(results) == {"knowledge_base_load", "single_scene_resolution", "full_pipeline"}
|
||||
assert all(value > 0 for value in results.values())
|
||||
|
||||
|
||||
def test_full_pipeline_is_fast() -> None:
|
||||
# Generous, non-flaky bound: the non-Analyzer pipeline is sub-millisecond in
|
||||
# practice; 100 ms leaves ample margin under CI load.
|
||||
kb = load_reference_knowledge_base()
|
||||
scene = SCENES["complex_multi_character"]
|
||||
assert measure_full_pipeline(kb, scene, iterations=20) < 0.1
|
||||
|
||||
|
||||
def test_recorded_baseline_exists() -> None:
|
||||
assert BASELINE_FILE.is_file()
|
||||
assert "Performance baselines" in BASELINE_FILE.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_knowledge_base_loaded_once_and_reused(tmp_path, monkeypatch) -> None:
|
||||
(tmp_path / "hair.json").write_text(
|
||||
'[{"id": "long_hair", "tags": ["long hair"], "category": "hair"}]',
|
||||
encoding="utf-8",
|
||||
)
|
||||
calls = {"n": 0}
|
||||
real_load = kb_module.load_knowledge_base
|
||||
|
||||
def counting_load(*args, **kwargs):
|
||||
calls["n"] += 1
|
||||
return real_load(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(kb_module, "load_knowledge_base", counting_load)
|
||||
|
||||
loader = kb_module.KnowledgeBaseLoader(tmp_path)
|
||||
loaded = [loader.get() for _ in range(5)]
|
||||
|
||||
assert calls["n"] == 1 # loaded once
|
||||
assert all(kb is loaded[0] for kb in loaded) # reused
|
||||
|
||||
|
||||
def test_pipeline_does_not_reload_knowledge_base(monkeypatch) -> None:
|
||||
kb = load_reference_knowledge_base()
|
||||
calls = {"n": 0}
|
||||
monkeypatch.setattr(
|
||||
kb_module, "load_knowledge_base", lambda *a, **k: calls.__setitem__("n", calls["n"] + 1)
|
||||
)
|
||||
for scene in SCENES.values():
|
||||
compile_prompt_outputs(scene, kb)
|
||||
assert calls["n"] == 0 # compilation reuses the passed Knowledge Base
|
||||
Reference in New Issue
Block a user