Text-to-SQL 数据库 AI 查询完全教程

适用读者：后端工程师、数据分析师、AI 应用开发者 预计阅读时间：25 分钟 最后更新：2026-05

目录

1. 什么是 Text-to-SQL
2. 技术原理与架构
3. 主流模型对比：SQLCoder vs Defog SQL
4. 多数据库支持与方言适配
5. Schema 理解与链接
6. 复杂查询处理
7. 准确性优化策略
8. 与数据分析 Agent 集成
9. 安全防护：SQL 注入防护
10. 企业级部署方案
11. 评测基准：Spider 与 BIRD
12. 最佳实践总结

1. 什么是 Text-to-SQL

Text-to-SQL（自然语言转 SQL）是将用户的自然语言问题自动转换为可执行 SQL 语句的技术。它让用户无需掌握 SQL 语法，直接用中文或英文提问即可查询数据库。

典型应用场景

与传统 BI 工具的区别

• 传统 BI：拖拽维度和指标，受限于预定义模型
• Text-to-SQL：自由提问，动态生成查询，灵活性极高
• 互补关系：Text-to-SQL 适合探索性查询，BI 适合固定报表

2. 技术原理与架构

2.1 整体架构

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  用户自然语言  │────▶│  预处理与意图  │────▶│  Schema 链接  │
│    输入        │     │    识别        │     │  & 表/列映射  │
└──────────────┘     └──────────────┘     └──────┬───────┘
                                                  │
                                                  ▼
┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  结果返回 &    │◀────│  SQL 验证 &   │◀────│  LLM 生成     │
│  可视化展示    │     │  自动修复      │     │  SQL 语句     │
└──────────────┘     └──────────────┘     └──────────────┘

2.2 核心流程详解

第一步：预处理

import re

def preprocess_question(question: str) -> dict:
    """预处理用户问题，提取关键信息"""
    # 移除多余空格和标点
    question = re.sub(r'\s+', ' ', question.strip())
    
    # 检测时间表达式
    time_patterns = {
        r'上个月': "DATE_SUB(CURDATE(), INTERVAL 1 MONTH)",
        r'最近7天': "DATE_SUB(CURDATE(), INTERVAL 7 DAY)",
        r'今年': "YEAR(CURDATE())",
    }
    
    time_refs = {}
    for pattern, sql_expr in time_patterns.items():
        if re.search(pattern, question):
            time_refs[pattern] = sql_expr
    
    # 检测聚合意图
    agg_keywords = {
        '最多': 'MAX', '最少': 'MIN', '平均': 'AVG',
        '总计': 'SUM', '数量': 'COUNT', '排名': 'ORDER BY'
    }
    
    detected_aggs = {k: v for k, v in agg_keywords.items() if k in question}
    
    return {
        'original': question,
        'time_refs': time_refs,
        'aggregations': detected_aggs,
        'is_likely_group_by': any(k in question for k in ['按', '每个', '分组', 'by'])
    }

第二步：Schema 链接（详见第 5 节）

第三步：SQL 生成

def generate_sql_prompt(schema: str, question: str, db_dialect: str = "MySQL") -> str:
    """构建 Text-to-SQL 的 Prompt"""
    prompt = f"""你是一个专业的 SQL 专家。请根据以下数据库结构和用户问题，生成准确的 {db_dialect} SQL 语句。

## 数据库结构
{schema}

## 规则
1. 只返回可执行的 SQL 语句，不要解释
2. 使用 {db_dialect} 方言
3. 确保 JOIN 条件正确
4. 处理可能的 NULL 值
5. 优先使用索引友好的写法

## 用户问题
{question}

## SQL:
"""
    return prompt

3. 主流模型对比：SQLCoder vs Defog SQL

3.1 SQLCoder

SQLCoder 是由 Defog 公司开源的 Text-to-SQL 专用模型系列，基于 CodeLlama 微调而来。

核心特性：

• 专注于 SQL 生成任务，准确率高于通用模型
• 支持 PostgreSQL、MySQL、BigQuery、Snowflake 等多方言
• 开源可本地部署，数据不出企业

# SQLCoder 本地部署示例
from transformers import AutoTokenizer, AutoModelForCausalLM

model_name = "defog/sqlcoder-7b-2"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",
    load_in_4bit=True  # 4-bit 量化，节省显存
)

def ask_sqlcoder(question: str, schema: str) -> str:
    prompt = f"""### Task
Generate a SQL query to answer the following question.

### Database Schema
{schema}

### Question
{question}

### SQL
```sql
"""
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
    outputs = model.generate(**inputs, max_new_tokens=256, temperature=0.1)
    result = tokenizer.decode(outputs[0], skip_special_tokens=True)
    # 提取 SQL 部分
    sql_start = result.find("```sql") + 6
    sql_end = result.find("```", sql_start)
    return result[sql_start:sql_end].strip() if sql_end > sql_start else result[sql_start:].strip()

3.2 Defog SQL

Defog 提供商业版和开源版两种模型，在 Spider 和 BIRD 基准上表现优异。

3.3 模型对比表

3.4 如何选择

• 数据敏感 → SQLCoder 或 Defog 本地部署
• 追求准确率 → SQLCoder-34B 或 GPT-4
• 成本敏感 → SQLCoder-7B + 验证层
• 快速原型 → GPT-4 API（开发阶段）

4. 多数据库支持与方言适配

4.1 SQL 方言差异

不同数据库的 SQL 方言存在显著差异，Text-to-SQL 系统必须正确处理：

4.2 方言适配层

class SQLDialectAdapter:
    """SQL 方言适配器"""
    
    DIALECTS = {
        'mysql': {
            'limit': 'LIMIT {n} OFFSET {offset}',
            'concat': 'CONCAT({args})',
            'now': 'NOW()',
            'string_quote': "'",
            'date_add': "DATE_ADD({date}, INTERVAL {n} {unit})",
        },
        'postgresql': {
            'limit': 'LIMIT {n} OFFSET {offset}',
            'concat': "({args})",  # 使用 || 运算符
            'now': 'NOW()',
            'string_quote': "'",
            'date_add': "{date} + INTERVAL '{n} {unit}'",
        },
        'sqlserver': {
            'limit': 'OFFSET {offset} ROWS FETCH NEXT {n} ROWS ONLY',
            'concat': 'CONCAT({args})',
            'now': 'GETDATE()',
            'string_quote': "'",
            'date_add': "DATEADD({unit}, {n}, {date})",
        }
    }
    
    def __init__(self, dialect: str = 'mysql'):
        self.dialect = dialect
        self.config = self.DIALECTS.get(dialect, self.DIALECTS['mysql'])
    
    def adapt_sql(self, sql: str, target_dialect: str) -> str:
        """将 SQL 从当前方言转换为目标方言"""
        if self.dialect == target_dialect:
            return sql
        
        # 转换 LIMIT 语法
        if self.dialect == 'mysql' and target_dialect == 'sqlserver':
            sql = self._mysql_to_sqlserver_limit(sql)
        
        # 转换日期函数
        sql = self._adapt_date_functions(sql, target_dialect)
        
        return sql
    
    def _mysql_to_sqlserver_limit(self, sql: str) -> str:
        import re
        match = re.search(r'LIMIT\s+(\d+)\s+OFFSET\s+(\d+)', sql, re.IGNORECASE)
        if match:
            n, offset = match.groups()
            sql = re.sub(
                r'LIMIT\s+\d+\s+OFFSET\s+\d+',
                f'OFFSET {offset} ROWS FETCH NEXT {n} ROWS ONLY',
                sql, flags=re.IGNORECASE
            )
        return sql

4.3 实际适配示例

# 用户问题："查询薪资最高的前5名员工"
# MySQL 输出：
# SELECT employee_name, salary FROM employees ORDER BY salary DESC LIMIT 5

# 通过适配器转换为 SQL Server：
adapter = SQLDialectAdapter('mysql')
sql_server_sql = adapter.adapt_sql(
    "SELECT employee_name, salary FROM employees ORDER BY salary DESC LIMIT 5",
    'sqlserver'
)
# 结果：SELECT employee_name, salary FROM employees ORDER BY salary DESC 
#       OFFSET 0 ROWS FETCH NEXT 5 ROWS ONLY

5. Schema 理解与链接

5.1 为什么 Schema 理解至关重要

Text-to-SQL 的核心难题不是生成语法正确的 SQL，而是理解用户问题与数据库结构之间的映射关系。这被称为 Schema Linking（模式链接）。

常见挑战：

• 用户说"销量"，数据库字段叫 quantity_sold
• 用户说"客户"，数据库表叫 customers 还是 clients？
• 同名字段在不同表中含义不同（如 name 在 employees 和 products 表中）

5.2 Schema 表示方法

def format_schema_for_prompt(db_schema: dict) -> str:
    """将数据库 schema 格式化为 LLM 可理解的文本"""
    lines = []
    
    for table_name, table_info in db_schema.items():
        lines.append(f"Table: {table_name}")
        if table_info.get('description'):
            lines.append(f"  Description: {table_info['description']}")
        lines.append("  Columns:")
        
        for col in table_info['columns']:
            col_desc = f"    - {col['name']} ({col['type']})"
            if col.get('primary_key'):
                col_desc += " [PRIMARY KEY]"
            if col.get('foreign_key'):
                col_desc += f" [FK -> {col['foreign_key']}]"
            if col.get('description'):
                col_desc += f" -- {col['description']}"
            lines.append(col_desc)
        
        if table_info.get('sample_values'):
            lines.append(f"  Sample values: {table_info['sample_values']}")
        lines.append("")
    
    return "\n".join(lines)


# 使用示例
schema = {
    "employees": {
        "description": "员工信息表",
        "columns": [
            {"name": "emp_id", "type": "INT", "primary_key": True},
            {"name": "emp_name", "type": "VARCHAR(100)", "description": "员工姓名"},
            {"name": "department_id", "type": "INT", "foreign_key": "departments.dept_id"},
            {"name": "salary", "type": "DECIMAL(10,2)", "description": "月薪（元）"},
            {"name": "hire_date", "type": "DATE", "description": "入职日期"},
        ],
        "sample_values": "salary: 5000-50000"
    },
    "departments": {
        "description": "部门信息表",
        "columns": [
            {"name": "dept_id", "type": "INT", "primary_key": True},
            {"name": "dept_name", "type": "VARCHAR(50)", "description": "部门名称"},
        ]
    }
}

print(format_schema_for_prompt(schema))

5.3 增强 Schema Linking 的技术

方法一：语义相似度匹配

from sentence_transformers import SentenceTransformer
import numpy as np

class SchemaLinker:
    """基于语义相似度的 Schema 链接"""
    
    def __init__(self):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.column_embeddings = {}
    
    def index_schema(self, db_schema: dict):
        """预计算所有列名和描述的嵌入向量"""
        texts = []
        keys = []
        
        for table, info in db_schema.items():
            for col in info['columns']:
                # 组合列名、表名和描述作为检索文本
                text = f"{table}.{col['name']}"
                if col.get('description'):
                    text += f" {col['description']}"
                texts.append(text)
                keys.append((table, col['name']))
        
        embeddings = self.model.encode(texts)
        for key, emb in zip(keys, embeddings):
            self.column_embeddings[key] = emb
    
    def link(self, question: str, top_k: int = 5) -> list:
        """将问题中的词语链接到最相关的列"""
        question_emb = self.model.encode([question])[0]
        
        scores = []
        for (table, col), emb in self.column_embeddings.items():
            score = np.dot(question_emb, emb) / (
                np.linalg.norm(question_emb) * np.linalg.norm(emb)
            )
            scores.append((table, col, float(score)))
        
        scores.sort(key=lambda x: x[2], reverse=True)
        return scores[:top_k]


# 使用
linker = SchemaLinker()
linker.index_schema(schema)
results = linker.link("哪个部门的平均工资最高？")
# 输出: [('employees', 'salary', 0.72), ('departments', 'dept_name', 0.68), ...]

方法二：别名词典

# 业务术语到数据库字段的映射
ALIAS_DICTIONARY = {
    "销量": ("sales", "quantity"),
    "销售额": ("sales", "amount"),
    "客户": ("customers", "cust_name"),
    "产品名": ("products", "product_name"),
    "入职时间": ("employees", "hire_date"),
    "工资": ("employees", "salary"),
    "部门": ("departments", "dept_name"),
}

def enrich_schema_with_aliases(schema_text: str) -> str:
    """在 Schema 描述中注入别名信息"""
    alias_lines = ["## 常用业务术语映射"]
    for term, (table, col) in ALIAS_DICTIONARY.items():
        alias_lines.append(f"  - '{term}' → {table}.{col}")
    
    return schema_text + "\n\n" + "\n".join(alias_lines)

6. 复杂查询处理

6.1 JOIN 查询

多表关联是 Text-to-SQL 最常见的难点之一。

# 用户问题："列出每个部门的员工数量和平均薪资"
# 
# 正确的 SQL（需要 JOIN）：
"""
SELECT 
    d.dept_name,
    COUNT(e.emp_id) AS employee_count,
    AVG(e.salary) AS avg_salary
FROM departments d
LEFT JOIN employees e ON d.dept_id = e.department_id
GROUP BY d.dept_name
ORDER BY avg_salary DESC;
"""

# 生成 Prompt 时，必须提供完整的外键关系
JOIN_GUIDE = """
## 表关联关系
- employees.department_id → departments.dept_id (多对一)
- orders.customer_id → customers.cust_id (多对一)
- order_items.order_id → orders.order_id (多对一)
- order_items.product_id → products.product_id (多对一)
"""

6.2 子查询处理

# 用户问题："查询薪资高于公司平均薪资的员工"
#
# 子查询写法：
"""
SELECT emp_name, salary
FROM employees
WHERE salary > (SELECT AVG(salary) FROM employees);
"""

# 窗口函数写法（更高效）：
"""
SELECT emp_name, salary
FROM (
    SELECT emp_name, salary,
           AVG(salary) OVER () AS avg_salary
    FROM employees
) t
WHERE salary > avg_salary;
"""

6.3 窗口函数

# 用户问题："每个部门薪资排名前3的员工"
#
# 生成的 SQL：
"""
SELECT dept_name, emp_name, salary, rn
FROM (
    SELECT 
        d.dept_name,
        e.emp_name,
        e.salary,
        ROW_NUMBER() OVER (
            PARTITION BY d.dept_id 
            ORDER BY e.salary DESC
        ) AS rn
    FROM employees e
    JOIN departments d ON e.department_id = d.dept_id
) ranked
WHERE rn <= 3
ORDER BY dept_name, rn;
"""

# 常用窗口函数模板
WINDOW_FUNCTION_TEMPLATES = {
    "排名": "ROW_NUMBER() OVER (PARTITION BY {group} ORDER BY {order})",
    "累计求和": "SUM({col}) OVER (PARTITION BY {group} ORDER BY {order} ROWS UNBOUNDED PRECEDING)",
    "同比": "({col} - LAG({col}, 12) OVER (ORDER BY {date})) / LAG({col}, 12) OVER (ORDER BY {date})",
    "占比": "{col} / SUM({col}) OVER (PARTITION BY {group}) * 100",
}

6.4 CTE（公用表表达式）

# 复杂问题："找出连续3个月销售额下降的产品"
# 
# 生成的 SQL（使用 CTE 提高可读性）：
"""
WITH monthly_sales AS (
    SELECT 
        product_id,
        DATE_FORMAT(sale_date, '%Y-%m') AS month,
        SUM(amount) AS total_sales
    FROM sales
    WHERE sale_date >= DATE_SUB(CURDATE(), INTERVAL 6 MONTH)
    GROUP BY product_id, DATE_FORMAT(sale_date, '%Y-%m')
),
sales_with_lag AS (
    SELECT 
        product_id,
        month,
        total_sales,
        LAG(total_sales, 1) OVER (PARTITION BY product_id ORDER BY month) AS prev_1,
        LAG(total_sales, 2) OVER (PARTITION BY product_id ORDER BY month) AS prev_2
    FROM monthly_sales
)
SELECT DISTINCT product_id
FROM sales_with_lag
WHERE total_sales < prev_1 AND prev_1 < prev_2;
"""

7. 准确性优化策略

7.1 多轮自校验（Self-Correction）

class SQLSelfCorrector:
    """SQL 自校验器：执行验证 + 自动修复"""
    
    def __init__(self, llm_client, db_connection):
        self.llm = llm_client
        self.db = db_connection
    
    def generate_and_verify(self, question: str, schema: str, max_retries: int = 3) -> str:
        sql = self.llm.generate_sql(question, schema)
        
        for attempt in range(max_retries):
            # 1. 语法检查
            error = self.check_syntax(sql)
            if error:
                sql = self.fix_syntax_error(sql, error, question, schema)
                continue
            
            # 2. 执行检查
            try:
                result = self.db.execute(sql)
                # 3. 结果合理性检查
                if self.validate_result(result, question):
                    return sql
                else:
                    sql = self.fix_logic_error(sql, result, question, schema)
            except Exception as e:
                sql = self.fix_execution_error(sql, str(e), question, schema)
        
        return sql  # 返回最后一次尝试
    
    def check_syntax(self, sql: str) -> str | None:
        """使用 SQLParser 检查语法"""
        try:
            # 使用 sqlparse 进行基本语法检查
            import sqlparse
            parsed = sqlparse.parse(sql)
            if not parsed or not parsed[0].tokens:
                return "Empty SQL statement"
            return None
        except Exception as e:
            return str(e)
    
    def fix_execution_error(self, sql: str, error: str, question: str, schema: str) -> str:
        """根据执行错误修复 SQL"""
        fix_prompt = f"""The following SQL query has an error. Please fix it.

Original question: {question}
Database schema: {schema}
SQL: {sql}
Error: {error}

Return only the fixed SQL, no explanation.
"""
        return self.llm.generate(fix_prompt)
    
    def validate_result(self, result, question: str) -> bool:
        """检查查询结果是否合理"""
        if result is None:
            return False
        if isinstance(result, list) and len(result) == 0:
            # 空结果可能是合理的，也可能是查询条件有误
            return True  # 默认接受空结果
        return True

7.2 Few-Shot 示例注入

FEW_SHOT_EXAMPLES = """
## 示例

### 问题：哪个城市的客户最多？
```sql
SELECT city, COUNT(*) AS customer_count
FROM customers
GROUP BY city
ORDER BY customer_count DESC
LIMIT 1;

问题：上个月订单金额超过1000元的客户姓名和订单号

SELECT c.cust_name, o.order_id, o.total_amount
FROM customers c
JOIN orders o ON c.cust_id = o.customer_id
WHERE o.order_date >= DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 MONTH), '%Y-%m-01')
  AND o.order_date < DATE_FORMAT(CURDATE(), '%Y-%m-01')
  AND o.total_amount > 1000;

问题：每个产品类别的销售额占总销售额的百分比

SELECT 
    p.category,
    SUM(s.amount) AS category_sales,
    ROUND(SUM(s.amount) * 100.0 / (SELECT SUM(amount) FROM sales), 2) AS pct
FROM sales s
JOIN products p ON s.product_id = p.product_id
GROUP BY p.category
ORDER BY category_sales DESC;

"""

### 7.3 投票机制（Ensemble）

```python
from collections import Counter

def ensemble_sql_generation(question: str, schema: str, models: list, n_samples: int = 3) -> str:
    """使用多个模型/多次采样进行投票"""
    all_sqls = []
    
    for model in models:
        for _ in range(n_samples):
            sql = model.generate_sql(question, schema, temperature=0.3)
            normalized = normalize_sql(sql)  # 标准化：去除空格、统一大小写
            all_sqls.append(normalized)
    
    # 投票选出最常见的 SQL
    counter = Counter(all_sqls)
    best_sql, count = counter.most_common(1)[0]
    
    print(f"最佳 SQL 出现 {count}/{len(all_sqls)} 次，置信度: {count/len(all_sqls):.1%}")
    return best_sql


def normalize_sql(sql: str) -> str:
    """标准化 SQL 以便比较"""
    import re
    sql = sql.strip().rstrip(';')
    sql = re.sub(r'\s+', ' ', sql)  # 合并空白
    sql = sql.lower()
    # 统一别名写法
    sql = sql.replace(' as ', ' ')
    return sql

8. 与数据分析 Agent 集成

8.1 Agent 架构设计

class DataAnalysisAgent:
    """数据分析 Agent：整合 Text-to-SQL 与数据可视化"""
    
    def __init__(self, sql_generator, db_connector, chart_generator):
        self.sql_gen = sql_generator
        self.db = db_connector
        self.chart = chart_generator
    
    async def analyze(self, question: str) -> dict:
        """端到端数据分析流程"""
        
        # 1. 生成 SQL
        sql = self.sql_gen.generate(question)
        
        # 2. 执行查询
        data = self.db.execute(sql)
        columns = self.db.get_columns()
        
        # 3. 智能可视化建议
        chart_type = self.suggest_chart_type(question, columns, data)
        
        # 4. 生成图表
        chart_config = self.chart.generate(data, columns, chart_type)
        
        # 5. 生成文字摘要
        summary = self.generate_summary(question, data, columns)
        
        return {
            'sql': sql,
            'data': data,
            'columns': columns,
            'chart': chart_config,
            'summary': summary,
            'row_count': len(data)
        }
    
    def suggest_chart_type(self, question: str, columns: list, data: list) -> str:
        """根据数据特征推荐图表类型"""
        question_lower = question.lower()
        
        # 基于关键词的推荐
        if any(kw in question_lower for kw in ['趋势', '变化', '走势']):
            return 'line'
        if any(kw in question_lower for kw in ['占比', '比例', '份额']):
            return 'pie'
        if any(kw in question_lower for kw in ['对比', '比较']):
            return 'bar'
        
        # 基于数据特征的推荐
        if len(data) <= 8 and len(columns) == 2:
            return 'pie'
        if any('date' in c.lower() or 'time' in c.lower() for c in columns):
            return 'line'
        
        return 'bar'  # 默认柱状图
    
    def generate_summary(self, question: str, data: list, columns: list) -> str:
        """生成数据分析摘要"""
        summary_parts = []
        
        summary_parts.append(f"**问题**: {question}")
        summary_parts.append(f"**返回 {len(data)} 条记录**")
        
        if len(data) > 0:
            # 尝试找出最大值、最小值等
            for i, col in enumerate(columns):
                values = [row[i] for row in data if row[i] is not None]
                if values and isinstance(values[0], (int, float)):
                    summary_parts.append(f"- {col}: 最小={min(values)}, 最大={max(values)}, 平均={sum(values)/len(values):.2f}")
        
        return "\n".join(summary_parts)

8.2 与 LangChain 集成

from langchain.agents import create_sql_agent
from langchain.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit

# 初始化数据库连接
db = SQLDatabase.from_uri("mysql://user:pass@localhost:3306/mydb")

# 创建 SQL Agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0)

toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent = create_sql_agent(
    llm=llm,
    toolkit=toolkit,
    verbose=True,
    agent_type="openai-functions"
)

# 使用
result = agent.invoke({"input": "哪个部门的离职率最高？"})
print(result['output'])

9. 安全防护：SQL 注入防护

9.1 Text-to-SQL 的安全风险

虽然 SQL 是由 LLM 生成而非用户直接输入，但仍存在严重安全风险：

1. 提示注入攻击：用户在问题中嵌入恶意指令
2. 权限越界：生成的 SQL 访问了不该访问的表
3. 数据泄露：通过精心构造的问题提取敏感数据
4. 破坏性操作：生成 DROP、DELETE、UPDATE 等危险语句

9.2 多层防护体系

import re
import sqlparse

class SQLSecurityGuard:
    """SQL 安全防护器"""
    
    # 黑名单关键词
    DANGEROUS_KEYWORDS = {
        'DROP', 'DELETE', 'TRUNCATE', 'ALTER', 'CREATE', 'INSERT',
        'UPDATE', 'GRANT', 'REVOKE', 'EXEC', 'EXECUTE', 'xp_',
        'sp_', 'INTO OUTFILE', 'INTO DUMPFILE', 'LOAD_FILE',
        'BENCHMARK', 'SLEEP', 'WAITFOR'
    }
    
    # 只允许 SELECT 语句
    ALLOWED_STATEMENTS = {'SELECT'}
    
    # 禁止访问的表（包含敏感数据）
    RESTRICTED_TABLES = {
        'user_credentials', 'api_keys', 'payment_cards',
        'audit_logs', 'system_config'
    }
    
    def validate(self, sql: str) -> tuple[bool, str]:
        """验证 SQL 安全性，返回 (是否通过, 原因)"""
        
        # 1. 基本格式检查
        if not sql or not sql.strip():
            return False, "空 SQL 语句"
        
        # 2. 解析 SQL
        parsed = sqlparse.parse(sql)
        if not parsed:
            return False, "无法解析 SQL"
        
        stmt = parsed[0]
        stmt_type = stmt.get_type()
        
        # 3. 只允许 SELECT
        if stmt_type not in self.ALLOWED_STATEMENTS:
            return False, f"禁止 {stmt_type} 操作，只允许 SELECT"
        
        sql_upper = sql.upper()
        
        # 4. 检查危险关键词
        for keyword in self.DANGEROUS_KEYWORDS:
            # 使用词边界匹配，避免误报
            if re.search(rf'\b{keyword}\b', sql_upper):
                return False, f"包含危险关键词: {keyword}"
        
        # 5. 检查多语句（防止堆叠注入）
        if ';' in sql.rstrip(';'):
            return False, "禁止多语句执行"
        
        # 6. 检查注释（可能隐藏恶意代码）
        if '--' in sql or '/*' in sql:
            return False, "禁止 SQL 注释"
        
        # 7. 检查受限表
        for table in self.RESTRICTED_TABLES:
            if table.upper() in sql_upper:
                return False, f"无权访问表: {table}"
        
        # 8. 检查子查询深度（防止 DoS）
        subquery_count = sql_upper.count('SELECT') - 1
        if subquery_count > 3:
            return False, f"子查询层数过多 ({subquery_count + 1}层)"
        
        return True, "通过"
    
    def sanitize_and_limit(self, sql: str, max_rows: int = 1000) -> str:
        """清理 SQL 并添加行数限制"""
        sql = sql.strip().rstrip(';')
        
        # 自动添加 LIMIT（防止全表扫描）
        if 'LIMIT' not in sql.upper():
            sql += f" LIMIT {max_rows}"
        
        return sql


# 使用示例
guard = SQLSecurityGuard()

# 正常查询 - 通过
ok, reason = guard.validate("SELECT name FROM employees WHERE dept = 'IT'")
print(f"通过: {ok}, 原因: {reason}")  # 通过: True, 原因: 通过

# 攻击尝试 - 拦截
ok, reason = guard.validate("SELECT * FROM employees; DROP TABLE employees;")
print(f"通过: {ok}, 原因: {reason}")  # 通过: False, 原因: 禁止多语句执行

ok, reason = guard.validate("SELECT * FROM user_credentials")
print(f"通过: {ok}, 原因: {reason}")  # 通过: False, 原因: 无权访问表: user_credentials

9.3 最小权限数据库用户

-- 为 Text-to-SQL 系统创建专用只读用户
CREATE USER 'text2sql_readonly'@'%' IDENTIFIED BY 'strong_password';

-- 只授予 SELECT 权限
GRANT SELECT ON mydb.employees TO 'text2sql_readonly'@'%';
GRANT SELECT ON mydb.departments TO 'text2sql_readonly'@'%';
GRANT SELECT ON mydb.sales TO 'text2sql_readonly'@'%';
-- 不授予敏感表权限

-- 设置查询超时
SET GLOBAL max_execution_time = 5000;  -- 5秒超时

-- 设置结果集大小限制
SET GLOBAL sql_select_limit = 10000;

10. 企业级部署方案

10.1 系统架构

                        ┌─────────────────┐
                        │   负载均衡器      │
                        └────────┬────────┘
                                 │
              ┌──────────────────┼──────────────────┐
              │                  │                  │
        ┌─────▼─────┐    ┌──────▼─────┐    ┌──────▼─────┐
        │ API 服务 1  │    │ API 服务 2  │    │ API 服务 3  │
        └─────┬─────┘    └──────┬─────┘    └──────┬─────┘
              │                  │                  │
              └──────────────────┼──────────────────┘
                                 │
                    ┌────────────┼────────────┐
                    │            │            │
              ┌─────▼─────┐ ┌───▼────┐ ┌────▼─────┐
              │ LLM 推理   │ │ Redis  │ │ PostgreSQL│
              │ 集群       │ │ 缓存   │ │ 查询日志  │
              └───────────┘ └────────┘ └──────────┘

10.2 高性能 API 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from functools import lru_cache
import hashlib
import json

app = FastAPI(title="Text-to-SQL API")

class QueryRequest(BaseModel):
    question: str
    database: str = "default"
    dialect: str = "mysql"
    max_rows: int = 100

class QueryResponse(BaseModel):
    sql: str
    data: list
    columns: list
    row_count: int
    execution_time_ms: float
    cached: bool = False

# SQL 缓存（基于问题哈希）
sql_cache = {}

@app.post("/query", response_model=QueryResponse)
async def text_to_sql_query(req: QueryRequest):
    # 1. 生成缓存键
    cache_key = hashlib.md5(
        f"{req.question}:{req.database}:{req.dialect}".encode()
    ).hexdigest()
    
    # 2. 检查缓存
    if cache_key in sql_cache:
        cached = sql_cache[cache_key]
        return QueryResponse(**cached, cached=True)
    
    # 3. 安全检查
    guard = SQLSecurityGuard()
    
    # 4. 生成 SQL
    sql = await generate_sql_async(req.question, req.database, req.dialect)
    
    # 5. 安全验证
    is_safe, reason = guard.validate(sql)
    if not is_safe:
        raise HTTPException(status_code=400, detail=f"安全检查失败: {reason}")
    
    # 6. 添加行数限制
    sql = guard.sanitize_and_limit(sql, req.max_rows)
    
    # 7. 执行查询
    import time
    start = time.time()
    data, columns = await execute_query_async(sql, req.database)
    exec_time = (time.time() - start) * 1000
    
    # 8. 记录日志
    await log_query(req.question, sql, len(data), exec_time)
    
    # 9. 缓存结果
    result = {
        'sql': sql, 'data': data, 'columns': columns,
        'row_count': len(data), 'execution_time_ms': round(exec_time, 2)
    }
    sql_cache[cache_key] = result
    
    return QueryResponse(**result)


# 异步 SQL 生成
async def generate_sql_async(question: str, database: str, dialect: str) -> str:
    schema = get_schema(database)
    prompt = generate_sql_prompt(schema, question, dialect)
    # 调用 LLM
    return await llm_client.generate(prompt)


# 查询审计日志
async def log_query(question: str, sql: str, row_count: int, exec_time: float):
    log_entry = {
        'timestamp': datetime.utcnow().isoformat(),
        'question': question,
        'generated_sql': sql,
        'row_count': row_count,
        'execution_time_ms': exec_time,
    }
    # 写入审计日志（用于后续分析和优化）
    await audit_logger.log(log_entry)

10.3 监控指标

# Prometheus 指标
from prometheus_client import Counter, Histogram, Gauge

QUERY_COUNT = Counter(
    'text2sql_queries_total', 
    'Total queries', 
    ['database', 'dialect', 'status']
)

QUERY_LATENCY = Histogram(
    'text2sql_query_latency_seconds',
    'Query latency',
    ['database', 'stage']  # stage: generate, execute, total
)

SQL_CACHE_HITS = Counter(
    'text2sql_cache_hits_total',
    'Cache hits'
)

ACTIVE_CONNECTIONS = Gauge(
    'text2sql_active_connections',
    'Active database connections'
)

11. 评测基准：Spider 与 BIRD

11.1 Spider 基准

Spider 是最广泛使用的 Text-to-SQL 评测基准，由耶鲁大学于 2018 年发布。

数据集特点：

• 包含 10,181 个问题，覆盖 200 个数据库
• 训练集和测试集的数据库完全不重叠（测试泛化能力）
• 分为简单、中等、困难三个难度级别

评测指标：

• Exact Match (EM)：生成的 SQL 与标准答案完全匹配
• Execution Accuracy (EX)：生成的 SQL 执行结果与标准答案一致
• Valid SQL (V)：生成的 SQL 语法正确可执行

11.2 BIRD 基准

BIRD（BIg Bench for LaRge-scale Database Grounded Text-to-SQL）是 2023 年发布的更贴近真实场景的基准。

与 Spider 的区别：

11.3 如何在 BIRD 上评估

import json
import sqlite3

def evaluate_on_bird(predictions: dict, bird_data_path: str) -> dict:
    """在 BIRD 基准上评估模型"""
    
    # 加载 BIRD 数据
    with open(f"{bird_data_path}/dev.json") as f:
        dev_data = json.load(f)
    
    results = {
        'total': len(dev_data),
        'correct': 0,
        'valid': 0,
        'by_difficulty': {'simple': [0, 0], 'moderate': [0, 0], 'challenging': [0, 0]}
    }
    
    for item in dev_data:
        qid = item['question_id']
        db_id = item['db_id']
        gold_sql = item['SQL']
        difficulty = item.get('difficulty', 'moderate')
        
        if qid not in predictions:
            continue
        
        pred_sql = predictions[qid]
        
        # 执行准确性检查
        try:
            db_path = f"{bird_data_path}/dev_databases/{db_id}/{db_id}.sqlite"
            conn = sqlite3.connect(db_path)
            
            gold_result = set(conn.execute(gold_sql).fetchall())
            pred_result = set(conn.execute(pred_sql).fetchall())
            
            results['valid'] += 1
            
            if gold_result == pred_result:
                results['correct'] += 1
                results['by_difficulty'][difficulty][0] += 1
            
            results['by_difficulty'][difficulty][1] += 1
            conn.close()
        except Exception:
            pass  # 无效 SQL
    
    # 计算准确率
    results['accuracy'] = results['correct'] / results['total'] if results['total'] > 0 else 0
    results['validity'] = results['valid'] / results['total'] if results['total'] > 0 else 0
    
    for diff in results['by_difficulty']:
        correct, total = results['by_difficulty'][diff]
        results['by_difficulty'][diff] = {
            'correct': correct,
            'total': total,
            'accuracy': correct / total if total > 0 else 0
        }
    
    return results

11.4 当前 SOTA 排行（截至 2026 年）

注意：以上数据为公开可查的基准结果，实际部署中准确率受 Schema 质量、Prompt 设计等因素影响。

12. 最佳实践总结

12.1 开发阶段

1. Schema 文档化：为每个表和关键字段添加描述，这是提升准确率最有效的手段
2. 建立别名词典：收集业务常用术语到数据库字段的映射
3. 准备测试用例：覆盖简单查询、多表 JOIN、聚合、子查询、窗口函数等场景
4. 渐进式上线：先在内部数据分析师群体试用，收集反馈

12.2 生产阶段

1. 强制安全层：所有生成的 SQL 必须经过安全校验
2. 只读权限：数据库用户只授予 SELECT 权限
3. 查询日志：记录所有生成的 SQL，用于审计和优化
4. 超时控制：设置查询超时，防止慢查询影响数据库
5. 结果缓存：相同问题短时间内直接返回缓存结果

12.3 持续优化

1. 收集 Bad Case：定期分析执行失败或结果不正确的查询
2. 更新 Few-Shot 示例：将典型的正确查询加入 Prompt
3. 微调模型：积累足够数据后，在自己的业务场景上微调专用模型
4. A/B 测试：对比不同 Prompt、不同模型的实际效果

12.4 常见陷阱

参考资源

• Defog SQLCoder GitHub
• Spider 数据集
• BIRD 数据集
• LangChain SQL Agent 文档
• SQLGlot - SQL 方言转换器

下一篇推荐阅读：AI应用性能优化与成本控制完全教程