Spaces:
Sleeping
Sleeping
File size: 5,175 Bytes
4f25e4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """Configuration comparison: run e2e eval across multiple pipeline settings.
Usage (from the rag-qa/ directory):
python -m evaluation.compare path/to/doc.pdf [more.pdf ...]
"""
import logging
import sys
from pathlib import Path
from ingestion.embedder import Embedder
from ingestion.pipeline import IngestionPipeline
from retrieval.index import VectorIndex
from generation.generator import Generator
from evaluation.eval_e2e import PLACEHOLDER_TEST_CASES, evaluate_e2e
logger = logging.getLogger(__name__)
_DEFAULT_CONFIGURATIONS = [
{"label": "baseline (no rerank)", "chunk_size": 500, "overlap": 50, "use_reranking": False},
{"label": "with reranking", "chunk_size": 500, "overlap": 50, "use_reranking": True},
{"label": "larger chunks + rerank","chunk_size": 1000, "overlap": 100, "use_reranking": True},
]
def compare_configurations(
pdf_paths: list[str | Path],
test_cases: list[dict] | None = None,
configurations: list[dict] | None = None,
rate_limit_delay: float = 1.5,
) -> list[dict]:
"""Re-ingest PDFs under multiple configurations and compare end-to-end metrics.
Each configuration dict supports:
label (str) — display name (auto-generated if omitted)
chunk_size (int) — default 500
overlap (int) — default 50
use_reranking (bool) — default True
Args:
pdf_paths: PDFs to ingest for each configuration.
test_cases: Labelled queries (defaults to PLACEHOLDER_TEST_CASES).
configurations: List of config dicts (defaults to three standard presets).
rate_limit_delay: Seconds between Gemini judge calls.
Returns:
List of result dicts with label and aggregate metric values. Also prints
a formatted comparison table.
"""
if test_cases is None:
test_cases = PLACEHOLDER_TEST_CASES
if configurations is None:
configurations = _DEFAULT_CONFIGURATIONS
pdf_paths = [Path(p) for p in pdf_paths]
missing = [p for p in pdf_paths if not p.exists()]
if missing:
raise FileNotFoundError(f"PDF(s) not found: {missing}")
print("Loading embedder and generator (shared across all configurations)...")
embedder = Embedder()
generator = Generator()
rows: list[dict] = []
for idx, cfg in enumerate(configurations, start=1):
label = cfg.get("label") or f"config-{idx}"
chunk_size = int(cfg.get("chunk_size", 500))
overlap = int(cfg.get("overlap", 50))
use_reranking = bool(cfg.get("use_reranking", True))
print(f"\n[{idx}/{len(configurations)}] {label} "
f"(chunk={chunk_size}, overlap={overlap}, rerank={use_reranking})")
index = VectorIndex(dimension=embedder.dimension)
pipeline = IngestionPipeline(
embedder=embedder, index=index,
strategy="recursive_character",
chunk_size=chunk_size, overlap=overlap,
)
for p in pdf_paths:
r = pipeline.ingest_pdf(p)
if r.error:
logger.warning(" Skipped %s: %s", p.name, r.error)
else:
print(f" {p.name}: {r.chunks} chunks")
metrics = evaluate_e2e(
test_cases, embedder, index, generator,
rate_limit_delay=rate_limit_delay, use_reranking=use_reranking,
)
print(f" → {metrics}")
rows.append({
"label": label,
"chunk_size": chunk_size,
"overlap": overlap,
"use_reranking": use_reranking,
"hit_rate": metrics.retrieval_hit_rate,
"faithfulness": metrics.avg_faithfulness,
"relevance": metrics.avg_relevance,
"keyword_cov": metrics.avg_keyword_coverage,
"composite": metrics.avg_composite,
})
_print_table(rows)
return rows
def _print_table(rows: list[dict]) -> None:
L, H, F, R, K, C = 28, 7, 8, 8, 10, 10
total = L + H + F + R + K + C
div = "-" * total
print(f"\n{'Configuration Comparison':^{total}}")
print(div)
print(
"Configuration".ljust(L) + "Hit".ljust(H) + "Faith.".ljust(F)
+ "Relev.".ljust(R) + "Keywords".ljust(K) + "Composite".ljust(C)
)
print(div)
for row in rows:
label = (row["label"][:L - 2] + "…") if len(row["label"]) > L - 1 else row["label"]
print(
label.ljust(L)
+ f"{row['hit_rate']:.2f}".ljust(H)
+ f"{row['faithfulness']:.2f}".ljust(F)
+ f"{row['relevance']:.2f}".ljust(R)
+ f"{row['keyword_cov']:.2f}".ljust(K)
+ f"{row['composite']:.2f}".ljust(C)
)
print(div)
best = max(rows, key=lambda r: r["composite"])
print(f"\nBest by composite score: {best['label']} ({best['composite']:.2f})")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(name)s | %(message)s")
if len(sys.argv) < 2:
print("Usage: python -m evaluation.compare <path/to/doc.pdf> [more.pdf ...]")
sys.exit(0)
compare_configurations([Path(p) for p in sys.argv[1:]])
|