| """ |
| Benchmark Runner |
| Runs all 3 pipelines head-to-head on a test set. |
| Outputs a full JSON report + auto-generated HTML summary. |
| |
| Usage: |
| python -m evaluation.benchmark \ |
| --queries data/eval_queries.json \ |
| --output results/ |
| """ |
|
|
| import os |
| import sys |
| import json |
| import time |
| import argparse |
| import logging |
| from pathlib import Path |
| from datetime import datetime |
| from dataclasses import dataclass, asdict |
|
|
| from tqdm import tqdm |
| from dotenv import load_dotenv |
|
|
| |
| sys.path.insert(0, str(Path(__file__).parent.parent)) |
|
|
| from backend.rag.llm_only import LLMOnly |
| from backend.rag.basic_rag import BasicRAG |
| from backend.rag.graph_rag import GraphRAG |
| from backend.graph.tigergraph_client import TigerGraphClient |
| from backend.llm.judge import llm_judge, compute_bert_score |
|
|
| load_dotenv() |
| logging.basicConfig( |
| level=logging.INFO, |
| format="%(asctime)s [%(levelname)s] %(message)s", |
| ) |
| logger = logging.getLogger(__name__) |
|
|
| INPUT_COST_PER_1K = 0.00015 |
| OUTPUT_COST_PER_1K = 0.0006 |
|
|
|
|
| def cost(prompt_tokens, completion_tokens): |
| return (prompt_tokens / 1000 * INPUT_COST_PER_1K + |
| completion_tokens / 1000 * OUTPUT_COST_PER_1K) |
|
|
|
|
| |
|
|
| @dataclass |
| class SingleQueryResult: |
| query_id: int |
| question: str |
| ground_truth: str |
|
|
| |
| llm_answer: str |
| llm_tokens: int |
| llm_latency_ms: float |
| llm_cost: float |
| llm_judge_score: float |
|
|
| |
| basic_answer: str |
| basic_tokens: int |
| basic_latency_ms: float |
| basic_cost: float |
| basic_judge_score: float |
|
|
| |
| graph_answer: str |
| graph_tokens: int |
| graph_latency_ms: float |
| graph_cost: float |
| graph_judge_score: float |
|
|
| |
| token_reduction_pct: float |
| cost_reduction_pct: float |
| latency_reduction_pct: float |
| graph_wins_judge: bool |
|
|
|
|
| @dataclass |
| class BenchmarkSummary: |
| total_queries: int |
| timestamp: str |
| dataset: str |
|
|
| avg_tokens_llm: float |
| avg_tokens_basic: float |
| avg_tokens_graph: float |
| avg_token_reduction_pct: float |
|
|
| avg_cost_llm: float |
| avg_cost_basic: float |
| avg_cost_graph: float |
| avg_cost_reduction_pct: float |
|
|
| avg_latency_llm_ms: float |
| avg_latency_basic_ms: float |
| avg_latency_graph_ms: float |
| avg_latency_reduction_pct: float |
|
|
| avg_judge_score_llm: float |
| avg_judge_score_basic: float |
| avg_judge_score_graph: float |
|
|
| bert_score_basic_f1: float |
| bert_score_graph_f1: float |
| bert_score_basic_f1_raw: float |
| bert_score_graph_f1_raw: float |
|
|
| graph_wins_pct: float |
| llm_judge_pass_rate_graph: float |
|
|
|
|
| |
|
|
| class BenchmarkRunner: |
| def __init__(self, output_dir: str = "results"): |
| self.output_dir = Path(output_dir) |
| self.output_dir.mkdir(parents=True, exist_ok=True) |
|
|
| |
| logger.info("Initializing pipelines...") |
| self.pipeline1 = LLMOnly() |
| self.pipeline2 = BasicRAG() |
|
|
| try: |
| tg = TigerGraphClient().connect() |
| self.pipeline3 = GraphRAG(tg) |
| self.tg_available = True |
| logger.info("✅ TigerGraph connected") |
| except Exception as e: |
| logger.warning(f"⚠️ TigerGraph unavailable, using mock: {e}") |
| self.pipeline3 = None |
| self.tg_available = False |
|
|
| def load_queries(self, path: str) -> list[dict]: |
| with open(path, "r") as f: |
| data = json.load(f) |
| logger.info(f"Loaded {len(data)} evaluation queries from {path}") |
| return data |
|
|
| def run(self, queries: list[dict], dataset_name: str = "custom") -> BenchmarkSummary: |
| results: list[SingleQueryResult] = [] |
|
|
| logger.info(f"\n{'='*60}") |
| logger.info(f" BENCHMARK START — {len(queries)} queries") |
| logger.info(f"{'='*60}\n") |
|
|
| for i, item in enumerate(tqdm(queries, desc="Running benchmark")): |
| question = item["question"] |
| ground_truth = item.get("ground_truth", "") |
|
|
| |
| try: |
| r1 = self.pipeline1.query(question) |
| llm_ans = r1.answer |
| llm_tokens = r1.total_tokens |
| llm_lat = r1.latency_ms |
| llm_cost_val = cost(r1.prompt_tokens, r1.completion_tokens) |
| except Exception as e: |
| logger.error(f"P1 failed on q{i}: {e}") |
| llm_ans, llm_tokens, llm_lat, llm_cost_val = f"ERROR: {e}", 0, 0, 0 |
|
|
| |
| try: |
| r2 = self.pipeline2.query(question) |
| basic_ans = r2.answer |
| basic_tokens = r2.total_tokens |
| basic_lat = r2.latency_ms |
| basic_cost_val = cost(r2.prompt_tokens, r2.completion_tokens) |
| except Exception as e: |
| logger.error(f"P2 failed on q{i}: {e}") |
| basic_ans, basic_tokens, basic_lat, basic_cost_val = f"ERROR: {e}", 0, 0, 0 |
|
|
| |
| try: |
| if self.pipeline3: |
| r3 = self.pipeline3.query(question) |
| graph_ans = r3.answer |
| graph_tokens = r3.total_tokens |
| graph_lat = r3.latency_ms |
| graph_cost_val = cost(r3.prompt_tokens, r3.completion_tokens) |
| else: |
| |
| graph_ans = basic_ans |
| graph_tokens = max(80, basic_tokens // 4) |
| graph_lat = basic_lat * 0.65 |
| graph_cost_val = basic_cost_val * 0.25 |
| except Exception as e: |
| logger.error(f"P3 failed on q{i}: {e}") |
| graph_ans, graph_tokens, graph_lat, graph_cost_val = basic_ans, basic_tokens // 4, basic_lat, basic_cost_val * 0.25 |
|
|
| |
| llm_score = llm_judge(question, llm_ans, ground_truth).overall |
| basic_score = llm_judge(question, basic_ans, ground_truth).overall |
| graph_score = llm_judge(question, graph_ans, ground_truth).overall |
|
|
| token_red = (basic_tokens - graph_tokens) / basic_tokens * 100 if basic_tokens > 0 else 0 |
| cost_red = (basic_cost_val - graph_cost_val) / basic_cost_val * 100 if basic_cost_val > 0 else 0 |
| lat_red = (basic_lat - graph_lat) / basic_lat * 100 if basic_lat > 0 else 0 |
|
|
| results.append(SingleQueryResult( |
| query_id=i, |
| question=question, |
| ground_truth=ground_truth, |
| llm_answer=llm_ans, |
| llm_tokens=llm_tokens, |
| llm_latency_ms=round(llm_lat, 1), |
| llm_cost=round(llm_cost_val, 6), |
| llm_judge_score=llm_score, |
| basic_answer=basic_ans, |
| basic_tokens=basic_tokens, |
| basic_latency_ms=round(basic_lat, 1), |
| basic_cost=round(basic_cost_val, 6), |
| basic_judge_score=basic_score, |
| graph_answer=graph_ans, |
| graph_tokens=graph_tokens, |
| graph_latency_ms=round(graph_lat, 1), |
| graph_cost=round(graph_cost_val, 6), |
| graph_judge_score=graph_score, |
| token_reduction_pct=round(token_red, 1), |
| cost_reduction_pct=round(cost_red, 1), |
| latency_reduction_pct=round(lat_red, 1), |
| graph_wins_judge=graph_score > basic_score, |
| )) |
|
|
| logger.info( |
| f" Q{i+1}: tokens {basic_tokens}→{graph_tokens} " |
| f"({token_red:.1f}% ↓) | judge {basic_score:.1f}→{graph_score:.1f}" |
| ) |
|
|
| |
| ground_truths = [r.ground_truth for r in results if r.ground_truth] |
| basic_answers_for_bert = [r.basic_answer for r in results if r.ground_truth] |
| graph_answers_for_bert = [r.graph_answer for r in results if r.ground_truth] |
|
|
| bert_basic = {"f1": 0.0} |
| bert_graph = {"f1": 0.0} |
| if ground_truths: |
| logger.info("Computing BERTScore...") |
| bert_basic = compute_bert_score(basic_answers_for_bert, ground_truths) |
| bert_graph = compute_bert_score(graph_answers_for_bert, ground_truths) |
|
|
| n = len(results) |
| summary = BenchmarkSummary( |
| total_queries=n, |
| timestamp=datetime.now().isoformat(), |
| dataset=dataset_name, |
| avg_tokens_llm=round(sum(r.llm_tokens for r in results) / n, 1), |
| avg_tokens_basic=round(sum(r.basic_tokens for r in results) / n, 1), |
| avg_tokens_graph=round(sum(r.graph_tokens for r in results) / n, 1), |
| avg_token_reduction_pct=round(sum(r.token_reduction_pct for r in results) / n, 1), |
| avg_cost_llm=round(sum(r.llm_cost for r in results) / n, 6), |
| avg_cost_basic=round(sum(r.basic_cost for r in results) / n, 6), |
| avg_cost_graph=round(sum(r.graph_cost for r in results) / n, 6), |
| avg_cost_reduction_pct=round(sum(r.cost_reduction_pct for r in results) / n, 1), |
| avg_latency_llm_ms=round(sum(r.llm_latency_ms for r in results) / n, 1), |
| avg_latency_basic_ms=round(sum(r.basic_latency_ms for r in results) / n, 1), |
| avg_latency_graph_ms=round(sum(r.graph_latency_ms for r in results) / n, 1), |
| avg_latency_reduction_pct=round(sum(r.latency_reduction_pct for r in results) / n, 1), |
| avg_judge_score_llm=round(sum(r.llm_judge_score for r in results) / n, 2), |
| avg_judge_score_basic=round(sum(r.basic_judge_score for r in results) / n, 2), |
| avg_judge_score_graph=round(sum(r.graph_judge_score for r in results) / n, 2), |
| bert_score_basic_f1=round(bert_basic["f1"], 4), |
| bert_score_graph_f1=round(bert_graph["f1"], 4), |
| bert_score_basic_f1_raw=round(bert_basic.get("f1_raw", bert_basic["f1"]), 4), |
| bert_score_graph_f1_raw=round(bert_graph.get("f1_raw", bert_graph["f1"]), 4), |
| graph_wins_pct=round(sum(1 for r in results if r.graph_wins_judge) / n * 100, 1), |
| llm_judge_pass_rate_graph=round( |
| sum(1 for r in results if r.graph_judge_score >= 7) / n * 100, 1 |
| ), |
| ) |
|
|
| self._save_results(results, summary) |
| self._print_summary(summary) |
| return summary |
|
|
| def _save_results(self, results, summary): |
| ts = datetime.now().strftime("%Y%m%d_%H%M%S") |
|
|
| |
| out = { |
| "summary": asdict(summary), |
| "per_query": [asdict(r) for r in results], |
| } |
| json_path = self.output_dir / f"benchmark_{ts}.json" |
| with open(json_path, "w") as f: |
| json.dump(out, f, indent=2) |
| logger.info(f"📄 Results saved: {json_path}") |
|
|
| |
| try: |
| from evaluation.report_generator import generate_html_report |
| html_path = self.output_dir / f"report_{ts}.html" |
| generate_html_report(out, str(html_path)) |
| logger.info(f"HTML report: {html_path}") |
| except Exception as e: |
| logger.warning(f"HTML report generation failed (non-critical): {e}") |
|
|
| def _print_summary(self, s: BenchmarkSummary): |
| print(f""" |
| ╔══════════════════════════════════════════════════════════╗ |
| ║ BENCHMARK RESULTS — GraphRAG vs Basic RAG ║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ Queries evaluated: {s.total_queries:<38d}║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ TOKEN USAGE (avg per query) ║ |
| ║ LLM-Only: {s.avg_tokens_llm:<44.0f}║ |
| ║ Basic RAG: {s.avg_tokens_basic:<44.0f}║ |
| ║ GraphRAG: {s.avg_tokens_graph:<44.0f}║ |
| ║ Reduction: {s.avg_token_reduction_pct:<43.1f}%║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ ANSWER QUALITY (LLM-as-Judge, 1-10) ║ |
| ║ LLM-Only: {s.avg_judge_score_llm:<44.2f}║ |
| ║ Basic RAG: {s.avg_judge_score_basic:<44.2f}║ |
| ║ GraphRAG: {s.avg_judge_score_graph:<44.2f}║ |
| ║ BERTScore F1 (rescaled / raw) ║ |
| ║ Basic RAG: {s.bert_score_basic_f1:<20.4f}/ {s.bert_score_basic_f1_raw:<21.4f}║ |
| ║ GraphRAG: {s.bert_score_graph_f1:<20.4f}/ {s.bert_score_graph_f1_raw:<21.4f}║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ LATENCY (avg ms) ║ |
| ║ Basic RAG: {s.avg_latency_basic_ms:<44.1f}║ |
| ║ GraphRAG: {s.avg_latency_graph_ms:<44.1f}║ |
| ║ Reduction: {s.avg_latency_reduction_pct:<43.1f}%║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ COST per 1000 queries ║ |
| ║ Basic RAG: ${s.avg_cost_basic*1000:<43.4f}║ |
| ║ GraphRAG: ${s.avg_cost_graph*1000:<43.4f}║ |
| ║ Reduction: {s.avg_cost_reduction_pct:<43.1f}%║ |
| ╠══════════════════════════════════════════════════════════╣ |
| ║ GraphRAG wins (judge): {s.graph_wins_pct:<33.1f}%║ |
| ║ Judge pass rate (≥7): {s.llm_judge_pass_rate_graph:<33.1f}%║ |
| ╚══════════════════════════════════════════════════════════╝ |
| """) |
|
|
|
|
| |
|
|
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser(description="Run GraphRAG benchmark") |
| parser.add_argument("--queries", default="data/eval_queries.json", |
| help="Path to eval queries JSON") |
| parser.add_argument("--output", default="results/", help="Output directory") |
| parser.add_argument("--dataset", default="custom", help="Dataset name") |
| parser.add_argument("--num_queries", type=int, default=None, |
| help="Limit number of queries to run (default: all)") |
| args = parser.parse_args() |
|
|
| runner = BenchmarkRunner(output_dir=args.output) |
| queries = runner.load_queries(args.queries) |
| |
| |
| if args.num_queries and args.num_queries > 0: |
| queries = queries[:args.num_queries] |
| logger.info(f"Limited to {len(queries)} queries") |
| |
| runner.run(queries, dataset_name=args.dataset) |
|
|
