Building a Production‑Ready SQL Evaluation Engine with Grok

Why You Need an Evaluation Engine for Text‑to‑SQL

Every time I ask a language model to translate a natural‑language request into SQL, the first thing that comes back is a candidate query.

If you’re building a product that powers analytics dashboards, billing reports or ad‑tech queries, a single wrong join can cost millions, and a missing filter could expose sensitive data.

I spent months sifting through hundreds of generated queries to find subtle bugs—wrong aggregation, omitted columns, or even the dreaded Cartesian product. The solution? A two‑layer evaluation framework that combines fast deterministic checks with an AI judge that explains why something is wrong and how to fix it.

Below I’ll walk you through the core ideas, show you the production‑ready code (no dashboards or storage needed), and explain how you can plug this into your existing workflow.

TL;DR – Build a deterministic 80/20 checker + an LLM “judge” that returns JSON with missing elements, root causes, and a corrected SQL. Run it asynchronously over hundreds of queries for real‑time quality control.

The Architecture in a Nutshell

User Query → Expected SQL (ground truth) 
            │
            ▼
LLM‑generated SQL ──► 1️⃣ Deterministic Evaluator  
                         │
                         ▼
                 Pass? ──> Yes → Return scores only  
                       No  → Trigger 2️⃣ LLM Judge

Layer 1 – Fast, deterministic checks that run in < 1 s per query.
- Row‑count match, column coverage, basic AST similarity, filter presence, aggregation fidelity.
Layer 2 – AI‑driven “judge” that:
- Compares the two SQL strings with a chain‑of‑thought prompt.
- Produces a JSON object: semantic score, missing elements, root cause, suggested fix.

The whole thing is pure Python; you can drop it into any CI pipeline or batch process.

Layer 1 – The Deterministic Core (80/20)

Below is the minimal yet powerful implementation that covers all the high‑impact metrics:

# deterministic_evaluator.py
import pandas as pd
from sqlglot import parse_one, exp

def get_tables(ast: exp.Expression) -> set:
    """Return a set of table names referenced in an AST."""
    return {t.this for t in ast.find_all(exp.Table)}

def evaluate_80_20(
    expected_sql: str,
    llm_sql: str,
    expected_df: pd.DataFrame,
    llm_df: pd.DataFrame
) -> dict:
    scores = {}

    # 1️⃣ Row Count Match
    scores["row_count_match"] = len(expected_df) == len(llm_df)
    scores["row_delta"] = abs(len(expected_df) - len(llm_df))
    scores["row_delta_pct"] = scores["row_delta"] / max(1, len(expected_df))

    # 2️⃣ Column Coverage
    exp_cols, llm_cols = set(expected_df.columns), set(llm_df.columns)
    scores["column_coverage"] = len(exp_cols & llm_cols) / len(exp_cols) if exp_cols else 0
    scores["missing_columns"] = list(exp_cols - llm_cols)
    scores["extra_columns"]   = list(llm_cols - exp_cols)

    # 3️⃣ Result Set Match (order‑insensitive)
    try:
        exp_sorted = expected_df.sort_values(by=list(expected_df.columns)).reset_index(drop=True)
        llm_sorted  = llm_df.sort_values(  by=list(llm_df.columns)).reset_index(drop=True)
        scores["result_exact_match"] = exp_sorted.equals(llm_sorted)
    except Exception:
        scores["result_exact_match"] = False

    # 4️⃣ Basic Structural Match via AST
    try:
        ast_exp = parse_one(expected_sql, read="postgres")
        ast_llm = parse_one(llm_sql,   read="postgres")

        scores["tables_match"]   = get_tables(ast_exp) == get_tables(ast_llm)
        scores["where_present"]  = bool(ast_exp.args.get("where")) == bool(ast_llm.args.get("where"))
        scores["group_by_match"] = str(ast_exp.args.get("group")) == str(ast_llm.args.get("group"))
    except Exception:
        scores.update({
            "tables_match": False,
            "where_present": False,
            "group_by_match": False
        })

    # 5️⃣ Simple Filter Check (presence of WHERE)
    scores["filter_match"] = ("WHERE" in expected_sql.upper()) == ("WHERE" in llm_sql.upper())

    # 6️⃣ Aggregation Presence
    agg_funcs = {"COUNT", "SUM", "AVG", "MAX", "MIN"}
    scores["has_aggregation"] = any(func in expected_sql.upper() for func in agg_funcs)

    # Weighted overall score (you can tweak weights)
    weights = {
        "row_count_match": 0.25,
        "column_coverage": 0.20,
        "result_exact_match": 0.30,
        "tables_match": 0.10,
        "group_by_match": 0.05,
        "filter_match": 0.05
    }
    overall = sum(scores[k] * w for k, w in weights.items() if isinstance(scores[k], bool))
    scores["overall_score"] = round(overall, 2)
    scores["verdict"] = (
        "PASS" if overall >= 0.9 else
        "PARTIAL" if overall >= 0.7 else
        "FAIL"
    )
    return scores

Why this works

Row count & column coverage catch the most common bugs (missing filters, wrong joins).
AST checks ensure you’re selecting from the right tables and applying the same grouping logic.
The weighted score gives you a quick pass/fail flag that can be used to gate deeper analysis.

Layer 2 – The AI Judge

When the deterministic layer flags anything below your threshold, we hand the case over to an LLM acting as a “debugger.”

The prompt is deliberately structured: it receives the user query, schema, both SQLs, and sample result rows. It then outputs only JSON with:

Semantic score (0‑100).
Missing elements (e.g., filter: status='ACTIVE').
Extra or wrong elements.
Root cause hypothesis.
Suggested corrected SQL.

# llm_judge.py
import json
from litellm import acompletion

JUDGE_PROMPT = """
You are an expert SQL analyst and debugger with 15+ years of experience in enterprise data warehouses.

Task: Deeply compare the **Expected SQL** (ground truth) with the **LLM-generated SQL**.
User natural-language query: {user_query}

Database Schema (full DDL):
{schema_ddl}

Expected SQL (ground truth):
{expected_sql}

LLM-generated SQL:
{llm_sql}

Expected Result (first 10 rows as CSV):
{expected_sample_csv}

LLM Result (first 10 rows as CSV):
{llm_sample_csv}

Row count — Expected: {exp_rows} | LLM: {llm_rows}
Columns — Expected: {exp_cols} | LLM: {llm_cols}

Evaluate using this exact rubric. Think step-by-step.

1. **Semantic Correctness** (0-100): How well does the LLM SQL answer the user query compared to the expected SQL?
2. **What is MISSING?** List every important element present in Expected but absent/mis‑implemented in LLM SQL.
3. **What is EXTRA or WRONG?** List anything in LLM SQL that should not be there or is incorrect.
4. **Root Cause Hypothesis**: Why might the LLM have made these mistakes given the original prompt?
5. **Suggested Fix**: Provide a complete, corrected SQL that would pass all checks. Make minimal changes possible while fully matching the expected semantics.

Return ONLY valid JSON (no markdown):
{{
  "semantic_score": integer 0-100,
  "missing_elements": [...],
  "extra_or_wrong_elements": [...],
  "root_cause": "...",
  "suggested_improved_sql": "...",
  "overall_verdict": "PASS | PARTIAL | FAIL",
  "explanation": "concise paragraph"
}}
"""
async def run_judge(
    user_query: str,
    schema_ddl: str,
    expected_sql: str,
    llm_sql: str,
    exp_sample_csv: str,
    llm_sample_csv: str,
    exp_rows: int,
    llm_rows: int,
    exp_cols: str,
    llm_cols: str
) -> dict:
    prompt = JUDGE_PROMPT.format(
        user_query=user_query,
        schema_ddl=schema_ddl,
        expected_sql=expected_sql,
        llm_sql=llm_sql,
        expected_sample_csv=exp_sample_csv,
        llm_sample_csv=llm_sample_csv,
        exp_rows=exp_rows,
        llm_rows=llm_rows,
        exp_cols=exp_cols,
        llm_cols=llm_cols
    )
    response = await acompletion(
        model="claude-3-5-haiku-20241022",  # cheap yet powerful
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,  # deterministic
        max_tokens=800,
    )
    content = response.choices[0].message.content.strip()
    return json.loads(content)

Why this works

The prompt forces the LLM to think step‑by‑step and produce structured JSON, making downstream parsing trivial.
By feeding sample rows (just 10) you avoid leaking sensitive data while still giving context for the judge.
Using a cheap model keeps per‑query cost under $0.01–$0.02.

Putting It All Together – Batch Runner


python
# batch_evaluator.py
import asyncio
from deterministic_evaluator import evaluate_80_20
from llm_judge import run_judge

async def evaluate_case(case: dict, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        det_scores = evaluate_80_20(
            case["expected_sql"],
            case["llm_sql"],
            case["expected_df"],
            case["llm_df"]
        )

        if det_scores["overall_score"] >= 0.92:
            return {**case, **det_scores, "judge_review": None}

        # Trigger LLM judge
        judge = await run_judge(
            user_query=case["user_query"],
            schema_ddl=case["schema_ddl"],
            expected_sql=case["expected_sql"],
            llm_sql=case["llm_sql"],
            exp_sample_csv=case["expected_df"].head(10).to_csv(index=False),
            llm_sample_csv=case["llm_df"].head(10).to_csv(index=False),
            exp_rows=len(case["expected_df"]),
            llm_rows=len(case["llm_df"]),
            exp_cols=",".join(case["expected_df"].columns),
            llm_cols=",".

Why You Need an Evaluation Engine for Text‑to‑SQL

The Architecture in a Nutshell

Layer 1 – The Deterministic Core (80/20)

Layer 2 – The AI Judge

Putting It All Together – Batch Runner

Layer 1 – The Deterministic Core (80/20)

Layer 2 – The AI Judge