Spaces:
Sleeping
Sleeping
File size: 3,799 Bytes
71c851f 0731ede 71c851f 0731ede 71c851f 0731ede 71c851f 0731ede 71c851f 0731ede 71c851f 0731ede 71c851f | 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 | # src/scripts/evaluate_hf.py
# Evaluation script for Hugging Face SQL generation.
import json
from pathlib import Path
import pandas as pd
from src.database.db_manager import get_db_connection
from src.nl2sql.sql_agent import nl2sql_agent
TEST_CASES_PATH = Path("src/scripts/test_cases.json")
RESULTS_PATH = Path("hf_evaluation_results.json")
def _normalize_dataframe(dataframe: pd.DataFrame) -> pd.DataFrame:
# Normalize dataframe to ensure accurate comparison
normalized = dataframe.copy()
normalized.columns = [str(column).lower() for column in normalized.columns]
for column in normalized.columns:
normalized[column] = normalized[column].map(
lambda value: round(float(value), 6)
if isinstance(value, float)
else value
)
sort_columns = list(normalized.columns)
if sort_columns:
normalized = normalized.sort_values(by=sort_columns, kind="mergesort").reset_index(drop=True)
return normalized
# Compare generated SQL results with expected results
def compare_results(df_generated: pd.DataFrame, df_gold: pd.DataFrame) -> bool:
"""Compare generated and expected query results."""
if df_generated is None or df_gold is None:
return False
try:
normalized_generated = _normalize_dataframe(df_generated)
normalized_gold = _normalize_dataframe(df_gold)
return normalized_generated.equals(normalized_gold)
except Exception as error:
print(f"Error comparing results: {error}")
return False
def run_evaluation():
if not TEST_CASES_PATH.exists():
print(f"Error: Could not find test cases at {TEST_CASES_PATH}")
return
with TEST_CASES_PATH.open("r", encoding="utf-8") as handle:
test_cases = json.load(handle)
results = []
correct_count = 0
print(f"Running evaluation on {len(test_cases)} test cases...\n")
for case in test_cases:
question = case["question"]
print(f"Testing ID {case['id']}: {question[:50]}...")
# Implement agent to handle RAG retrieval and SQL generation
agent_response = nl2sql_agent(user_question=question)
generated_sql = agent_response.get("query", "")
connection = get_db_connection()
if connection is None:
raise RuntimeError("Unable to connect to the SQLite database.")
try:
df_generated = pd.read_sql_query(generated_sql, connection)
df_gold = pd.read_sql_query(case["gold_sql"], connection)
is_correct = compare_results(df_generated, df_gold)
if is_correct:
correct_count += 1
results.append(
{
"id": case["id"],
"question": question,
"status": "PASS" if is_correct else "FAIL",
"generated_sql": generated_sql,
"gold_sql": case["gold_sql"],
}
)
except Exception as error:
results.append(
{
"id": case["id"],
"question": question,
"status": "ERROR",
"generated_sql": generated_sql,
"gold_sql": case["gold_sql"],
"error": str(error),
}
)
finally:
connection.close()
accuracy = (correct_count / len(test_cases)) * 100 if test_cases else 0.0
print("\nEVALUATION COMPLETE")
print(f"Total Test Cases: {len(test_cases)}")
print(f"Correctly Generated SQL: {correct_count} / {len(test_cases)}")
print(f"Execution Accuracy: {accuracy:.2f}%")
with RESULTS_PATH.open("w", encoding="utf-8") as handle:
json.dump(results, handle, indent=4) |