Spaces:
Sleeping
Sleeping
| """Reproducibility manifest for one benchmark run. | |
| Owns exactly one responsibility: capturing what would need to be identical to | |
| reproduce a run's numbers -- git commit, the corpus/query-set versions and | |
| their content hashes, and the library/model versions that influence retrieval | |
| behavior (embedding model, chromadb version). Emits nothing: this is metadata | |
| assembled once at the start of a run and written to `manifest.json` alongside | |
| `operational.jsonl` / `quality.jsonl` / `summary.json`. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import subprocess | |
| from dataclasses import asdict, dataclass, field | |
| from pathlib import Path | |
| import chromadb | |
| from app.benchmarks.rag_observability.corpus import ( | |
| Corpus, | |
| QuerySet, | |
| corpus_manifest_fingerprint, | |
| query_set_fingerprint, | |
| ) | |
| # The embedder is a fixed, hardcoded ONNX model (see app/rag/embeddings.py) -- | |
| # not configurable at runtime, so this is a literal, not a lookup. | |
| EMBEDDING_MODEL = "chromadb-onnx-all-MiniLM-L6-v2" | |
| # Self-documenting caveat about what `cache_state="cold"` does and does not | |
| # mean in this runner's rows (see runner.py's module docstring for the full | |
| # explanation). Carried into `manifest.json`/`summary.json` verbatim so a | |
| # reader consuming only the machine-readable artifacts -- not the source -- | |
| # cannot misread `cache_state=cold` as a true process-cold-start measurement. | |
| CACHE_STATE_SEMANTICS = ( | |
| "cache_state='cold' means first touch of this query in this process, " | |
| "before that query's own warm-up repetition -- a same-process proxy, " | |
| "not a fresh-process/cold-OS-cache measurement. True process-level cold " | |
| "isolation is out of scope for a single harness process." | |
| ) | |
| def git_commit() -> str | None: | |
| """Best-effort current commit hash. Never raises -- a manifest field being | |
| unavailable (e.g. running outside a git checkout) must degrade to `None`, | |
| not fail the whole benchmark run.""" | |
| try: | |
| result = subprocess.run( | |
| ["git", "rev-parse", "HEAD"], | |
| capture_output=True, | |
| text=True, | |
| timeout=5, | |
| cwd=Path(__file__).resolve().parent, | |
| ) | |
| if result.returncode == 0: | |
| return result.stdout.strip() or None | |
| except Exception: | |
| pass | |
| return None | |
| def cognee_version() -> str | None: | |
| """Informational only -- this benchmark never calls Cognee (pure ChromaDB | |
| retrieval), but `ExperimentIdentity` carries the field, so record it as | |
| installed-but-unused rather than silently omitting it.""" | |
| try: | |
| import importlib.metadata | |
| return importlib.metadata.version("cognee") | |
| except Exception: | |
| return None | |
| class RunManifest: | |
| run_id: str | |
| experiment_id: str | |
| pipeline_version: str | |
| smoke: bool | |
| seed: int | |
| warm_repetitions: int | |
| cold_query_count: int | |
| created_at: str # UTC ISO8601 | |
| git_commit: str | None | |
| embedding_model: str | |
| chroma_version: str | |
| cognee_version: str | None | |
| corpus_version: str | |
| corpus_document_count: int | |
| corpus_manifest_fingerprint: str | |
| query_set_version: str | |
| query_count: int | |
| query_set_fingerprint: str | |
| cache_state_semantics: str = CACHE_STATE_SEMANTICS | |
| def to_dict(self) -> dict: | |
| return asdict(self) | |
| def build_manifest( | |
| *, | |
| run_id: str, | |
| experiment_id: str, | |
| pipeline_version: str, | |
| smoke: bool, | |
| seed: int, | |
| warm_repetitions: int, | |
| cold_query_count: int, | |
| created_at: str, | |
| corpus: Corpus, | |
| query_set: QuerySet, | |
| ) -> RunManifest: | |
| return RunManifest( | |
| run_id=run_id, | |
| experiment_id=experiment_id, | |
| pipeline_version=pipeline_version, | |
| smoke=smoke, | |
| seed=seed, | |
| warm_repetitions=warm_repetitions, | |
| cold_query_count=cold_query_count, | |
| created_at=created_at, | |
| git_commit=git_commit(), | |
| embedding_model=EMBEDDING_MODEL, | |
| chroma_version=chromadb.__version__, | |
| cognee_version=cognee_version(), | |
| corpus_version=corpus.version, | |
| corpus_document_count=len(corpus.documents), | |
| corpus_manifest_fingerprint=corpus_manifest_fingerprint(), | |
| query_set_version=query_set.version, | |
| query_count=len(query_set.queries), | |
| query_set_fingerprint=query_set_fingerprint(), | |
| cache_state_semantics=CACHE_STATE_SEMANTICS, | |
| ) | |
| def write_manifest(manifest: RunManifest, path: Path) -> None: | |
| Path(path).write_text(json.dumps(manifest.to_dict(), indent=2) + "\n", encoding="utf-8") | |