近年来，大模型（LLM）技术，如ChatGPT、Snowflake Cortext AI、Claude和Gemini，在SQL查询生成方面展现出惊人的能力，使得自然语言查询数据库成为可能。然而，生成的SQL是否真正准确？如何客观评估其准确率？ 本文将深入探讨如何使用黄金标准对比、结果集验证以及可选的LLM作为裁判的语义评估，来评估LLM生成的SQL查询的准确性，并提供一个可立即使用的Python项目，助力你构建可靠的、生产级别的AI数据分析系统。

准确率评估的重要性

将自然语言作为数据库接口已经不再是遥远的未来。无论是ChatGPT、GitHub Copilot，还是企业定制的AI助手，都让分析师和业务用户能够通过简单的提示词编写复杂的查询。

然而，盲目信任LLM生成的SQL，就像让一位缺乏经验的分析师在蒙眼的情况下执行生产环境查询一样，风险重重。不正确的查询可能导致：

• 对AI系统信任的降低
• 返回不完整、不准确或具有误导性的数据
• 在关键业务流程中静默失败

因为LLM本质上是概率模型，而非确定性模型。这意味着它们可能生成看起来合理的SQL，但实际上无法完全满足业务需求，与预期的模式不匹配，或者更糟糕的是，返回不正确或不完整的结果。

因此，评估SQL准确率是构建可靠的、生产级别的、使用LLM进行数据分析的系统的关键。量化每个查询的“正确”程度，识别边缘情况和模型盲点，跟踪LLM SQL生成性能的改进，都离不开准确率的评估。

SQL生成评估工作流程

评估SQL查询准确率的核心在于对比LLM生成的SQL与“黄金标准”SQL，并进行详细的结果集验证，甚至可以引入LLM进行语义评估。

主要评估模式包括：

• 完全匹配 (Exact Match)：生成的SQL结果与黄金标准SQL结果在行和列上必须100%匹配。
• 子集匹配 (Subset Match)：检查生成的SQL结果是否为黄金标准SQL结果的子集，这种方式在一些情况下更灵活，例如，当LLM额外返回了一些不影响核心结果的数据时。
• LLM-as-a-Judge：使用另一个大模型（例如GPT-4）来评估生成的SQL与问题的相关性，这对于更细致的评估非常有帮助，尤其是在结果集比较不足以区分优劣的情况下。

项目结构搭建

为了更好地理解和实践，我们先创建一个项目结构：

llm_sql_accuracy_evaluator/
├── data/
│   └── db.sqlite             # 示例SQLite数据库
├── golden_queries/
│   └── benchmark.json        # JSON文件，包含问题和黄金标准SQL
├── src/
│   ├── evaluator.py          # 执行和比较查询的核心逻辑
│   ├── sql_generator.py      # LLM SQL生成逻辑的占位符
│   └── judge.py              # 可选：LLM作为裁判的评估
├── notebook/
│   └── example_usage.ipynb   # Jupyter Notebook中的示例用法
├── requirements.txt
└── README.md

黄金标准SQL的建立

首先，我们需要准备一个包含“黄金标准”SQL的JSON文件。这些“黄金标准”SQL是针对特定自然语言问题的理想SQL查询。

假设我们有一个简单的SQLite数据库，其中包含一个名为employees的表：

CREATE TABLE employees (
    id INTEGER,
    employee_name TEXT,
    department TEXT,
    salary INTEGER
);

然后，我们定义黄金标准SQL，例如：

// golden_queries/benchmark.json
{
  "Q1": {
    "question": "获取所有薪水大于50000的员工",
    "sql": "SELECT * FROM employees WHERE salary > 50000"
  },
  "Q2": {
    "question": "列出HR部门的员工姓名",
    "sql": "SELECT employee_name FROM employees WHERE department = 'HR'"
  }
}

这些黄金标准SQL将作为评估生成的SQL的基础。

LLM SQL生成

在实际应用中，SQL将由大模型生成。为了方便测试，我们可以创建一个模拟的LLM SQL生成器。

# src/sql_generator.py
def mock_llm_sql_generator(question):
    if "salary" in question:
        return "SELECT employee_name, employee_id FROM employees WHERE salary > 50000"
    return "SELECT * FROM employees"

注意，这个模拟生成器简化了输出，只返回两列，而不是所有列。

执行查询并比较结果

接下来，我们需要执行黄金标准SQL和生成的SQL，并在同一数据库上进行比较。

# src/evaluator.py
import sqlite3
import pandas as pd

DB_PATH = 'data/db.sqlite'

def run_query(query):
    conn = sqlite3.connect(DB_PATH)
    return pd.read_sql_query(query, conn)

这将生成两个 Pandas DataFrames，方便进行比较。

行列匹配分数计算

计算行列匹配分数是评估准确率的关键步骤。我们可以定义以下函数：

# src/evaluator.py (continued)

def column_match_score(df1, df2):
    cols_1 = set(df1.columns)
    cols_2 = set(df2.columns)
    return len(cols_1 & cols_2) / len(cols_1 | cols_2) * 100

def row_match_score(df1, df2):
    set_1 = set(df1.apply(tuple, axis=1))
    set_2 = set(df2.apply(tuple, axis=1))
    return len(set_1 & set_2) / len(set_1 | set_2) * 100

def compute_accuracy(df_gold, df_gen):
    row_score = row_match_score(df_gold, df_gen)
    col_score = column_match_score(df_gold, df_gen)
    return round((row_score + col_score) / 2, 2)

column_match_score 函数计算列的匹配程度。它通过计算两个DataFrame共有列的数量与总列数的比率来评估。row_match_score 函数则计算行的匹配程度。它将每一行转换为一个元组，计算两个DataFrame共有行的数量与总行数的比率。compute_accuracy 函数将行列匹配分数平均，得到最终的准确率。

现在可以使用这些函数来评估任何SQL查询。

from src.evaluator import run_query, compute_accuracy
from src.sql_generator import mock_llm_sql_generator
import json

with open('golden_queries/benchmark.json') as f:
    gold = json.load(f)
    question = gold['Q1']['question']
    golden_sql = gold['Q1']['sql']

generated_sql = mock_llm_sql_generator(question)
df_gold = run_query(golden_sql)
df_gen = run_query(generated_sql)
accuracy = compute_accuracy(df_gold, df_gen)

print(f"Accuracy: {accuracy}%")

例如，输出可能是： Accuracy: 75.0%，这意味着完全行匹配 (100%)，部分列匹配 (50%)，平均得分为 75%。

LLM-as-a-Judge：语义分数

为了进行更细致的评估，可以使用大模型作为裁判来评估查询。

# src/judge.py
import openai

def llm_judge_score(question, generated_sql, golden_sql):
    prompt = f"""你是一名专业的SQL审核员。
问题: {question}
黄金标准SQL: {golden_sql}
生成的SQL: {generated_sql}
请根据生成的SQL满足问题意图的程度，将生成的SQL的准确性从0到100进行评分。"""

    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return float(response.choices[0].message['content'].strip())

在这种方法中，我们使用 OpenAI 的 GPT-4 模型来评估生成的 SQL。 该模型会收到一个包含问题、黄金标准 SQL 和生成的 SQL 的提示，并被要求根据生成的 SQL 满足问题意图的程度对生成的 SQL 的准确性进行评分。当结果集比较不足时，这对于主观或无模式环境特别有用。 需要配置OpenAI API Key。

扩展框架到其他数据工程框架

为了扩展这个框架到其他数据工程框架，以下是一些增强建议：

1. 支持Snowflake和多个后端

   抽象数据库层以支持Snowflake、Postgres、BigQuery等多个引擎。

   # db_connector.py
   def run_query(query, engine='sqlite', conn_params={}):
       if engine == 'sqlite':
           import sqlite3
           conn = sqlite3.connect(conn_params.get("path", "data/db.sqlite"))
       elif engine == 'snowflake':
           import snowflake.connector
           conn = snowflake.connector.connect(**conn_params)
       else:
           raise ValueError("Unsupported engine")
       df = pd.read_sql(query, conn)
       return df
   

   可以扩展conn_params以包含Snowflake字段：

   {
     "user": "...",
     "password": "...",
     "account": "...",
     "warehouse": "...",
     "database": "...",
     "schema": "..."
   }
   

2. LLM查询评估的测试套件

   支持批量评估多个查询对（黄金标准+生成）。

   创建test_cases.json：

   [
     {
       "question_id": "Q1",
       "generated_sql": "SELECT id FROM employees WHERE salary > 50000"
     },
     {
       "question_id": "Q2",
       "generated_sql": "SELECT * FROM employees WHERE department = 'HR'"
     }
   ]
   

   然后创建一个批量评估器：

   def evaluate_batch(test_cases_path):
       with open(test_cases_path) as f:
           cases = json.load(f)
       results = []
       for case in cases:
           accuracy = evaluate_query_accuracy(case['question_id'], case['generated_sql'])
           results.append({**case, "accuracy": accuracy})
       return pd.DataFrame(results)
   

3. 模式和语义比较

   检查结构相似性，而不仅仅是结果内容。

   思路：

   • 使用SQL解析器，如sqlparse或mo_sql_parsing，来提取选定的字段、条件、连接。
   • 比较查询组件：SELECT、FROM、WHERE、GROUP BY等。
   • 在AST（抽象语法树）之间应用基于令牌的相似性（例如，Levenshtein、Jaccard）。

4. 指标仪表板（可选的Streamlit应用程序）

   以交互方式可视化准确性结果。

   创建一个Streamlit仪表板：

   import streamlit as st
   import pandas as pd
   
   def evaluate_query_accuracy(question_id, generated_query):
       # 你的评估逻辑
       return 0.0  # 这里只是一个占位符
   
   def evaluate_batch(test_cases_path):
       with open(test_cases_path) as f:
           cases = json.load(f)
       results = []
       for case in cases:
           accuracy = evaluate_query_accuracy(case['question_id'], case['generated_sql'])
           results.append({**case, "question_id": case['question_id'], "generated_sql":case['generated_sql'], "accuracy": accuracy})
       return pd.DataFrame(results)
   
   
   df = evaluate_batch("test_cases.json")
   st.dataframe(df)
   st.bar_chart(df.set_index("question_id")["accuracy"])
   

   这使得随着时间的推移审查LLM性能非常有用。

5. 高级LLM-as-a-Judge配置

   让用户在评分风格之间进行选择：

   • “严格”（逻辑+结构）
   • “宽松”（仅意图匹配）
   • “可解释”（也返回解释）

   使用可选的配置文件：

   {
     "llm_judge_mode": "explainable",
     "model": "gpt-4",
     "temperature": 0.2
   }
   

   更新judge.py以从中读取并根据模式更改响应格式。

6. 置信度评分+用于可审计性的日志记录

   向每个评估添加元数据：

   • 查询执行成功/失败
   • 运行时错误（如果有）
   • 置信度（来自法官或模型不确定性）

   返回评估元数据，例如：

   {
     "question_id": "Q1",
     "accuracy": 75.0,
     "row_score": 100.0,
     "column_score": 50.0,
     "execution_success": true,
     "llm_confidence": 0.87,
     "errors": null
   }
   

   将这些日志存储到本地文件或Snowflake表中，以进行QA工作流程。

7. CI/CD和GitHub Actions集成

   在pull请求中使用GitHub Actions运行SQL准确性测试。

   创建.github/workflows/sql_eval.yml：

   name: SQL Accuracy Evaluation
   
   on:
     push:
       branches: [ main ]
     pull_request:
       branches: [ main ]
   
   jobs:
     test:
       runs-on: ubuntu-latest
       steps:
         - uses: actions/checkout@v2
         - name: Set up Python 3.x
           uses: actions/setup-python@v2
           with:
             python-version: '3.x'
         - name: Install dependencies
           run: pip install -r requirements.txt
         - name: Run evaluator
           run: python run_batch_eval.py
   

8. Snowflake特定的增强功能

   • 处理完全限定的对象名称（DB.SCHEMA.TABLE）
   • 验证查询执行的角色基于访问控制（RBAC）
   • 使用Snowflake的结果缓存来降低成本
   • 为长时间运行的查询添加重试和超时逻辑

9. CI/CD集成亮点

   GitHub Actions会在每次推送到主分支或拉取请求时触发。

   从requirements.txt安装依赖项。

   运行evaluate_batch()函数以根据黄金标准查询验证生成的SQL。

   直接在CI日志中输出准确性分数。

现在可以：

• 将项目上传到GitHub存储库。
• GitHub Actions将自动在PR上运行以验证SQL准确性。
• 如果需要，可以使用Slack警报，PR评论或测试阈值扩展此功能。

部署到GitHub

1. 创建一个新的存储库（不Fork）：

   导航到GitHub并登录到您的帐户。

   单击右上角的“+”图标，然后选择“新建存储库”。

   命名您的存储库（例如，llm-sql-accuracy-evaluator）并提供描述。

   选择可见性（公共或私有），然后单击“创建存储库”。

2. 初始化本地存储库：

   在您的本地计算机上，打开一个终端并导航到您的项目目录。

   初始化Git：

   git init
   

   添加所有文件：

   git add .
   

   提交更改：

   git commit -m "Initial commit"
   

3. 链接到GitHub存储库：

   添加远程存储库：

   git remote add origin https://github.com/yourusername/llm-sql-accuracy-evaluator.git
   

   推送更改：

   git push -u origin master
   

4. 设置GitHub Pages（文档可选）：

   如果您想托管文档或静态站点，可以使用GitHub Pages：

   转到GitHub上的存储库。

   单击“设置”>“Pages”。

   在“来源”下，选择要发布的分支和文件夹（例如，/docs）。

   单击“保存”。

5. 配置GitHub Actions（CI/CD可选）：

   要自动化测试或部署等任务：

   在您的存储库中，单击“Actions”。

   选择工作流程模板或设置自定义工作流程。

   将工作流程文件提交到.github/workflows/目录。

总结与展望

评估LLM生成的SQL查询不仅仅是语法检查，更是确保数据完整性、意图对齐和业务正确性的关键。通过结合精确的结果比较、灵活的子集逻辑和LLM驱动的语义评估，可以获得一个强大的框架，用于实时验证SQL质量。

无论是构建智能AI工作流、BI副驾驶，还是内部查询助手，这个框架都至关重要。 随着大模型技术的不断发展，SQL准确率评估将成为数据分析领域不可或缺的一环，助力我们构建更可靠、更智能的数据应用。拥抱准确率评估，才能真正释放LLM在数据分析领域的潜力。

参考文章中提供的GitHub仓库（LLM SQL Accuracy Evaluator），你可以动手尝试，并根据自己的数据环境进行调整。 欢迎贡献和改进！