#!/usr/bin/env python3

"""

Batch evaluation script for search results using the API.

Evaluates BM25 (Conj/Disj), Semantic, and Hybrid (Rerank) results.

"""

from dataclasses import dataclass

from pathlib import Path

import requests

from tqdm import tqdm

import tyro

from python.evaluation.evaluate import evaluate, EvalConfig

@dataclass

class BatchEvalConfig:

"""Configuration for batch evaluation."""

results_dir: Path = Path("results_2020")

"""Directory to save result files"""

api_url: str = "http://localhost:8080/api/search"

"""API endpoint URL"""

output_table: bool = True

"""Print summary table"""

query_file: Path = Path("data/msmarco-test2019-queries.tsv")

"""Path to query file"""

qrels_file: Path = Path("data/2019qrels-docs.txt")

"""Path to qrels file"""

dataset_file: Path = Path("data/msmarco-docs.tsv")

def load_queries(query_file: Path) -> list[tuple[str, str]]:

"""Load queries from TSV file."""

queries = []

with open(query_file, "r", encoding="utf-8") as f:

for line in f:

parts = line.strip().split("\t")

if len(parts) >= 2:

queries.append((parts[0], parts[1]))

return queries

def run_system(

system_name: str,

mode: str,

rerank: bool,

queries: list[tuple[str, str]],

output_file: Path,

api_url: str,

):

"""Run queries against the API and save results."""

print(f"Running {system_name}...")

with open(output_file, "w", encoding="utf-8") as f:

for qid, query in tqdm(queries, desc=system_name):

try:

params = {

"q": query,

"mode": mode,

"top_k": 100, # Retrieve top 100 for evaluation

"rerank": str(rerank).lower(),

"snippet": "false",

}

response = requests.get(api_url, params=params)

if response.status_code == 200:

data = response.json()

results = data.get("results", [])

for i, res in enumerate(results):

# Format: query_id Q0 doc_id rank score run_name

external_doc_id = res.get("raw_doc_id", str(res["doc_id"]))

score = res["score"]

rank = i + 1

f.write(

f"{qid}\tQ0\t{external_doc_id}\t{rank}\t{score}\t{system_name}\n"

)

else:

# print(f"Error querying {qid}: {response.status_code}")

pass

except Exception as e:

print(f"Exception for query {qid}: {e}")

def run_batch_evaluation(config: BatchEvalConfig) -> dict:

"""

Run evaluation on the dataset.

Args:

config: Batch evaluation configuration

Returns:

Dictionary mapping system name to metrics

"""

print("\n" + "=" * 70)

print(" SEARCH SYSTEMS EVALUATION (API)")

print("=" * 70)

# Ensure results directory exists

config.results_dir.mkdir(exist_ok=True)

# Dataset configuration

query_file = config.query_file

qrels_file = config.qrels_file

dataset_file = config.dataset_file

if not query_file.exists():

print(f"Error: Query file not found: {query_file}")

return {}

if not qrels_file.exists():

print(f"Error: Qrels file not found: {qrels_file}")

return {}

if not dataset_file.exists():

print(f"Error: Dataset file not found: {dataset_file}")

return {}

# Load queries

print(f"Loading queries from {query_file}...")

queries = load_queries(query_file)

print(f"Loaded {len(queries)} queries.")

# Define systems to evaluate

systems = [

{

"name": "BM25",

"file": "bm25.txt",

"mode": "conjunctive",

"rerank": False,

},

{

"name": "BM25 + Rerank",

"file": "bm25_rerank.txt",

"mode": "conjunctive",

"rerank": True,

},

{

"name": "Semantic Search",

"file": "semantic.txt",

"mode": "semantic",

"rerank": False,

},

{

"name": "Semantic + Rerank",

"file": "semantic_rerank.txt",

"mode": "semantic",

"rerank": True,

},

]

all_results = {}

for system in systems:

output_file = config.results_dir / system["file"]

# Run queries and generate result file

run_system(

system["name"],

system["mode"],

system["rerank"],

queries,

output_file,

config.api_url,

)

# Evaluate

print(f"Evaluating {system['name']}...")

try:

eval_config = EvalConfig(

qrels=qrels_file, results=output_file, use_map=False, verbose=False

)

metrics = evaluate(eval_config)

all_results[system["name"]] = metrics

except Exception as e:

print(f"Evaluation error for {system['name']}: {e}")

# Print summary table

if config.output_table and all_results:

print("\n" + "=" * 95)

print(" SUMMARY TABLE")

print("=" * 95)

print(

f"{'System':<30} {'MRR@10':<10} {'Recall@100':<12} {'NDCG@10':<10} {'NDCG@100':<11}"

)

print("-" * 95)

for name, metrics in all_results.items():

mrr = f"{metrics.get('mrr@10', 0.0):.4f}"

recall = f"{metrics.get('recall@100', 0.0):.4f}"

ndcg10 = (

f"{metrics.get('ndcg@10', 0.0):.4f}" if "ndcg@10" in metrics else "-"

)

ndcg100 = (

f"{metrics.get('ndcg@100', 0.0):.4f}" if "ndcg@100" in metrics else "-"

)

print(f"{name:<30} {mrr:<10} {recall:<12} {ndcg10:<10} {ndcg100:<11}")

print("=" * 95)

print()

return all_results

def main(config: BatchEvalConfig) -> None:

"""Main batch evaluation function."""

run_batch_evaluation(config)

if __name__ == "__main__":

tyro.cli(main)