Gemini-SQL2 深度实战：当大模型学会「读表写SQL」——从 Text-to-SQL 原理到生产级自然语言数据库查询系统的完全指南（2026）

2026年6月12日，Google Research 发布了 Gemini-SQL2，在 BIRD 基准测试上以 80.04% 的执行准确率登顶单模型榜单。这意味着大模型把自然语言翻译成可执行 SQL 的能力又上了一个台阶。但刷榜和实操是两码事——本文不满足于"新闻速递"，而是带你从 Text-to-SQL 的技术原理到工程落地，彻底搞懂这项技术。

一、为什么 Text-to-SQL 是企业级 AI 落地的关键钥匙

1.1 数据分析师的日常噩梦

你可能见过这样的场景：

• 业务人员：「帮我查一下上季度华东区域退货率最高的前10个SKU」
• 数据分析师打开 Navicat，面对 200+ 张表、3000+ 字段，开始写一个涉及5张表 JOIN 的 SQL
• 30分钟后写完了，跑了一下——语法错误
• 修复语法，再跑——跑了3分钟还没出结果
• 业务人员：「等太久了，我先手动看Excel了」

这不是段子，而是每天都在发生的事。根据某头部电商数据团队统计，一个资深分析师平均每天要写 15-20 个 SQL 查询，其中 60% 是重复性业务查询，只有 40% 需要真正的分析能力。而业务人员的平均等待时间是 45 分钟。

Text-to-SQL 的核心价值就在于此：把「人写 SQL」这个瓶颈变成「AI 写 SQL，人审结果」。

1.2 Text-to-SQL 不是新概念，但 2026 年才是真正的拐点

Text-to-SQL 的发展经历了三个阶段：

阶段一（2017-2020）：学术探索期

• WikiSQL 数据集发布（2017），开创了 Text-to-SQL 研究方向
• Spider 数据集（2018），引入了跨表查询
• 准确率在 60-70% 之间徘徊
• 模型以 Seq2SQL、SQLNet 等专用小模型为主

阶段二（2021-2024）：大模型冲击期

• GPT-3、Codex 等大模型出现，zero-shot Text-to-SQL 成为可能
• 但幻觉问题严重，生成的 SQL 经常引用不存在的表或字段
• 在 Spider 上的准确率提升到 85%+，但 Spider 的表结构远比真实业务简单

阶段三（2025-2026）：生产化突破期

• BIRD 基准成为新的金标准——95 个数据库、37 个专业领域、12751 组问答对、33.4GB 数据量
• 引入脏数据和外部知识需求，评测难度远超 Spider
• Gemini-SQL（77.14%）→ Gemini-SQL2（80.04%），专用模型开始超越通用模型
• 开源生态爆发：Vanna、PremSQL、APEX-SQL 等框架让开发者可以快速集成

1.3 BIRD vs Spider：为什么 BIRD 更接近真实世界

关键区别在于：BIRD 要求模型不仅要理解 SQL 语法，还要理解业务语义。 比如「去年销售额增长最快的产品线」——你需要知道「销售额」对应哪个字段（可能是 revenue、sales_amount、total_price 等），「增长最快」意味着需要同比计算，而「产品线」可能是一个需要 JOIN 才能关联的字段。

二、Gemini-SQL2 技术深度解析

2.1 从 Gemini-SQL 到 Gemini-SQL2 的演进

Gemini-SQL（2025年底发布）：

• 基座模型：Gemini 2.5 Pro
• 训练方式：多任务监督微调（Multi-task Supervised Fine-tuning）
• BIRD 单模型执行准确率：77.14%
• 创新：首次在 BIRD 单模型赛道超越传统 pipeline 方法

Gemini-SQL2（2026年6月发布）：

• 基座模型：Gemini 3.1 Pro（升级了 2 代基座模型）
• BIRD 单模型执行准确率：80.04%（提升 2.9 个百分点）
• 关键改进推测（基于公开信息和技术趋势分析）：
  1. 更强的 Schema Linking：Gemini 3.1 Pro 的推理能力增强，对表结构中字段与自然语言问句的映射更准确
  2. 外部知识融合：更好地处理需要领域知识的查询（如「VIP 客户」的定义）
  3. 复杂 SQL 生成：窗口函数、CTE、多层级子查询的生成质量提升
  4. 错误自纠正：可能引入了 SQL 验证和自我修正机制

2.2 Text-to-SQL 的核心技术挑战

要理解 Gemini-SQL2 为什么强，首先要知道这个任务到底难在哪：

挑战一：Schema Linking（模式链接）

用户问题：查一下今年退货金额超过1000元的客户名单

数据库表结构（部分）：
- orders(id, customer_id, order_date, total_amount, status)
- returns(id, order_id, return_date, refund_amount, reason)
- customers(id, name, email, phone, vip_level, register_date)

模型需要完成以下映射：

• 「今年」→ return_date >= '2026-01-01'
• 「退货金额」→ returns.refund_amount
• 「超过1000元」→ refund_amount > 1000
• 「客户名单」→ 需要关联 customers 表，输出 customers.name

这看似简单，但在真实业务中：

• 字段名可能是 amt、refund、r_amount 等各种缩写
• 「客户」可能是 user、member、client、buyer
• 「今年」需要知道数据的时间范围和当前日期
• 一张 200 字段的表里，模型需要精准定位到正确的字段

挑战二：SQL 语法正确性

大模型擅长「看起来像 SQL」的文本，但不一定生成可执行的 SQL：

-- 大模型常见的错误模式

-- 错误1：表名/字段名不存在
SELECT cust_name FROM orders  -- orders 表没有 cust_name

-- 错误2：JOIN 条件错误
SELECT o.*, c.name
FROM orders o
JOIN customers c ON o.customer = c.id  -- 应该是 o.customer_id

-- 错误3：聚合逻辑错误
SELECT product, SUM(price) / COUNT(*) AS avg_price
FROM orders
GROUP BY product  -- SUM(price)/COUNT(*) 不等于平均值，应该用 AVG(price)

-- 错误4：子查询嵌套过深导致性能灾难
SELECT * FROM orders WHERE customer_id IN (
  SELECT customer_id FROM returns WHERE refund_amount > (
    SELECT AVG(refund_amount) * 2 FROM returns
  )
)
-- 可能导致全表扫描

挑战三：安全边界

这是最致命的挑战。 生产环境中，AI 生成的 SQL 如果不加限制：

-- 危险操作1：误删数据
DELETE FROM orders WHERE status = 'cancelled'

-- 危险操作2：全表更新
UPDATE customers SET vip_level = 1

-- 危险操作3：超大规模查询导致数据库宕机
SELECT * FROM orders o1
CROSS JOIN orders o2  -- 笛卡尔积！

-- 危险操作4：信息泄露
SELECT credit_card_number, cvv FROM customers

AI 不仅要学会写 SQL，更要学会「不写什么 SQL」。

2.3 Gemini-SQL2 的技术架构推测

虽然 Google 尚未发布技术报告，但根据 Gemini 3.1 Pro 的已知能力和 Text-to-SQL 领域的前沿研究，我们可以合理推测 Gemini-SQL2 的架构：

┌─────────────────────────────────────────────┐
│              用户自然语言输入                  │
│  "查一下今年Q1各区域的退货率和同比变化"           │
└────────────────────┬────────────────────────┘
                     ▼
┌─────────────────────────────────────────────┐
│           Schema Understanding Layer         │
│  1. 读取数据库 DDL（表结构、字段类型、注释）     │
│  2. 识别关键词映射：区域→region/area/district  │
│  3. 构建语义图：表关系、字段依赖、业务含义       │
│  4. 外部知识检索：业务术语词典、历史查询模式       │
└────────────────────┬────────────────────────┘
                     ▼
┌─────────────────────────────────────────────┐
│           Query Planning Layer               │
│  1. 意图分解：退货率 + 同比变化 + 按区域分组     │
│  2. SQL 模板选择：需要 CTE + 窗口函数          │
│  3. 列映射确认：确保每个引用的字段存在           │
│  4. 安全检查：只读确认、行数限制、超时预算         │
└────────────────────┬────────────────────────┘
                     ▼
┌─────────────────────────────────────────────┐
│           SQL Generation Layer (Gemini 3.1)  │
│  1. 基于增强 Prompt 生成 SQL                  │
│  2. 语法验证（AST 解析）                       │
│  3. 语义验证（执行计划分析）                    │
│  4. 自我修正循环（最多 N 轮）                   │
└────────────────────┬────────────────────────┘
                     ▼
┌─────────────────────────────────────────────┐
│           Execution & Result Layer            │
│  1. 在沙箱/只读副本上执行                      │
│  2. 结果格式化与解释                           │
│  3. 生成自然语言回答                           │
└─────────────────────────────────────────────┘

三、从零搭建生产级 Text-to-SQL 系统

3.1 技术选型对比

3.2 方案一：用 Vanna 快速搭建 Text-to-SQL 系统

Vanna 是目前最成熟的 Text-to-SQL 开源框架，基于 RAG 架构，支持 20+ 种数据库。

安装：

pip install vanna[anthropic]
# 或者用 OpenAI
pip install vanna[openai]
# 本地模型
pip install vanna[ollama]

快速上手——5分钟搭一个自然语言查询系统：

from vanna import Agent
from vanna.integrations.anthropic import AnthropicLlmService
from vanna.core.registry import ToolRegistry
from vanna.tools import DatabaseTools
import sqlite3

# 第一步：初始化 Agent
agent = Agent(
    llm_service=AnthropicLlmService(
        model="claude-sonnet-4-20250514",
        api_key="your-api-key"
    ),
    registry=ToolRegistry()
)

# 第二步：注册数据库工具
db = sqlite3.connect("your_database.db")
db_tools = DatabaseTools(db)
agent.registry.register_tool(db_tools)

# 第三步：训练——让 AI 学习你的数据库结构
# 训练 DDL（表结构定义）
agent.train_ddl("""
CREATE TABLE customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT,
    phone TEXT,
    vip_level INTEGER DEFAULT 0,
    city TEXT,
    register_date DATE
);

CREATE TABLE orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(id),
    order_date DATE,
    total_amount REAL,
    status TEXT CHECK(status IN ('pending','completed','cancelled','refunded')),
    region TEXT
);

CREATE TABLE order_items (
    id INTEGER PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    product_id INTEGER,
    quantity INTEGER,
    unit_price REAL
);

CREATE TABLE returns (
    id INTEGER PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    return_date DATE,
    refund_amount REAL,
    reason TEXT
);
""")

# 训练业务文档（让 AI 理解业务术语）
agent.train_documentation("""
业务术语说明：
- VIP客户：vip_level >= 2 的客户
- 退货率：退货订单数 / 总订单数
- Q1：1月-3月，Q2：4月-6月，Q3：7月-9月，Q4：10月-12月
- 活跃客户：最近90天有下单记录的客户
""")

# 训练示例 SQL（提供正确示例供 AI 参考）
agent.train_sql(
    question="查上个月的销售额",
    sql="SELECT SUM(total_amount) FROM orders WHERE order_date >= date('now', '-1 month')"
)
agent.train_sql(
    question="各城市的订单数量",
    sql="SELECT c.city, COUNT(o.id) FROM orders o JOIN customers c ON o.customer_id = c.id GROUP BY c.city ORDER BY COUNT(o.id) DESC"
)

# 第四步：开始查询
question = "今年Q1各区域的退货率是多少？"
sql = agent.generate_sql(question)
print(f"生成的 SQL：{sql}")

result = db.execute(sql).fetchall()
print(f"查询结果：{result}")

Vanna 的 RAG 工作原理：

训练阶段：
  DDL/文档/示例SQL → 向量化 → 存入向量数据库（如 ChromaDB）

查询阶段：
  用户问题 → 向量化 → 检索相关DDL/文档/SQL → 构造增强Prompt → LLM生成SQL

这种架构的核心优势是可解释性：你可以查看 AI 检索了哪些 DDL 和示例 SQL 来生成答案，方便调优和排错。

3.3 方案二：用 LangChain + DeepSeek 构建自有系统

如果你需要更精细的控制，可以基于 LangChain 自建：

from langchain_community.utilities import SQLDatabase
from langchain_experimental.sql import SQLDatabaseChain
from langchain_community.chat_models import ChatOllama
from langchain.prompts import PromptTemplate

# 连接数据库
db = SQLDatabase.from_uri("postgresql://user:pass@localhost/mydb")

# 获取表结构信息
tables_info = db.get_table_info()
print(f"数据库包含 {len(db.get_usable_table_names())} 张表")

# 自定义 Prompt——这是提升准确率的关键
SQL_PROMPT = PromptTemplate.from_template("""
你是一个 SQL 专家。根据以下数据库信息和用户问题，生成准确的 SQL 查询。

## 数据库表结构
{table_info}

## 用户问题
{input}

## 规则
1. 只使用 SELECT 语句，禁止 INSERT/UPDATE/DELETE/DROP
2. 优先使用 EXISTS 子查询而非 IN，提升性能
3. 大表查询必须带 WHERE 条件和时间范围限制
4. 对金额计算使用 DECIMAL 类型，避免浮点误差
5. 只返回 SQL，不要解释

## SQL
""")

# 使用本地模型（Ollama + DeepSeek-Coder）
llm = ChatOllama(model="deepseek-coder-v2", temperature=0)

# 创建 Chain
db_chain = SQLDatabaseChain.from_llm(
    llm,
    db,
    prompt=SQL_PROMPT,
    verbose=True,  # 打印中间过程，方便调试
    return_direct=True  # 返回 SQL 而非执行结果
)

# 查询
question = "列出今年退货金额最高的10个客户及其退货总金额"
sql = db_chain.invoke({"query": question})
print(f"生成的 SQL：{sql}")

3.4 方案三：企业级 Text-to-SQL 架构（推荐用于生产环境）

"""
企业级 Text-to-SQL 服务架构
特点：SQL 安全检查、结果缓存、审计日志、限流保护
"""
import re
import hashlib
import time
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import sqlparse
from sqlparse.sql import Where, Parenthesis
from sqlparse.tokens import Keyword, DML


class SQLSafetyLevel(Enum):
    SAFE = "safe"
    WARNING = "warning"
    DANGEROUS = "dangerous"
    FORBIDDEN = "forbidden"


@dataclass
class SQLCheckResult:
    level: SQLSafetyLevel
    issues: list[str]
    suggestions: list[str]
    estimated_rows: Optional[int] = None
    estimated_cost: Optional[float] = None


class SQLSafetyChecker:
    """SQL 安全检查器——生产环境的最后一道防线"""

    # 禁止的操作
    FORBIDDEN_KEYWORDS = [
        'DROP', 'TRUNCATE', 'ALTER', 'CREATE', 'GRANT', 'REVOKE',
        'INSERT', 'UPDATE', 'DELETE', 'REPLACE'
    ]

    # 敏感字段（不允许 SELECT）
    SENSITIVE_FIELDS = [
        'password', 'credit_card', 'cvv', 'ssn', 'id_card',
        'bank_account', 'secret_key', 'token'
    ]

    # 大表配置（超过此行数需要加 LIMIT）
    LARGE_TABLES = {
        'orders': 10_000_000,
        'order_items': 50_000_000,
        'logs': 100_000_000,
    }

    def check(self, sql: str, schema: dict) -> SQLCheckResult:
        """执行完整的安全检查"""
        issues = []
        suggestions = []
        level = SQLSafetyLevel.SAFE

        sql_upper = sql.upper().strip()

        # 检查1：禁止的操作
        for keyword in self.FORBIDDEN_KEYWORDS:
            if re.search(rf'\b{keyword}\b', sql_upper):
                return SQLCheckResult(
                    level=SQLSafetyLevel.FORBIDDEN,
                    issues=[f"禁止使用 {keyword} 操作"],
                    suggestions=["此操作在只读模式下不可用"]
                )

        # 检查2：是否以 SELECT 开头
        if not sql_upper.startswith('SELECT'):
            return SQLCheckResult(
                level=SQLSafetyLevel.FORBIDDEN,
                issues=["只允许 SELECT 查询"],
                suggestions=["请修改为只读查询"]
            )

        # 检查3：敏感字段访问
        for field in self.SENSITIVE_FIELDS:
            if field in sql_upper:
                level = SQLSafetyLevel.FORBIDDEN
                issues.append(f"不允许查询敏感字段: {field}")

        # 检查4：LIMIT 检查（防止全表扫描）
        parsed = sqlparse.parse(sql)[0]
        has_limit = any(
            token.match(sqlparse.tokens.Keyword, 'LIMIT')
            for token in parsed.flatten()
        )
        if not has_limit:
            # 检查是否涉及大表
            for table, threshold in self.LARGE_TABLES.items():
                if table in sql_upper:
                    level = SQLSafetyLevel.WARNING
                    issues.append(f"查询涉及大表 {table}（预估 {threshold:,} 行）")
                    suggestions.append("建议添加 LIMIT 子句限制返回行数")
                    break

        # 检查5：CROSS JOIN 检测
        if 'CROSS JOIN' in sql_upper:
            level = SQLSafetyLevel.DANGEROUS
            issues.append("检测到 CROSS JOIN，可能导致笛卡尔积")
            suggestions.append("请确认是否需要 CROSS JOIN，通常应使用 INNER JOIN 替代")

        # 检查6：子查询深度
        paren_count = sql.count('(')
        if paren_count > 5:
            level = SQLSafetyLevel.WARNING
            issues.append(f"子查询嵌套层级较深（{paren_count} 层）")
            suggestions.append("考虑使用 CTE (WITH 子句) 替代多层嵌套")

        # 检查7：WHERE 子句检查
        where_found = any(isinstance(token, Where) for token in parsed.flatten())
        if not where_found and 'LIMIT' not in sql_upper:
            level = SQLSafetyLevel.WARNING
            issues.append("查询缺少 WHERE 条件")
            suggestions.append("无条件的查询可能导致全表扫描，建议添加 WHERE 限制")

        return SQLCheckResult(
            level=level,
            issues=issues,
            suggestions=suggestions
        )


class TextToSQLService:
    """Text-to-SQL 核心服务"""

    def __init__(self, db_uri: str, llm_client, schema_store):
        self.db = SQLDatabase.from_uri(db_uri)
        self.llm = llm_client
        self.schema_store = schema_store
        self.safety_checker = SQLSafetyChecker()
        self.cache = {}  # 简化的查询缓存
        self.audit_log = []  # 审计日志

    def generate_sql(self, question: str, user_id: str) -> dict:
        """生成 SQL 的完整流程"""
        start_time = time.time()

        # 步骤1：检查缓存
        cache_key = hashlib.md5(
            f"{question}:{user_id}".encode()
        ).hexdigest()
        if cache_key in self.cache:
            return {**self.cache[cache_key], "cached": True}

        # 步骤2：增强 Prompt（注入 Schema 信息）
        schema_info = self.schema_store.get_relevant_schema(question)
        enhanced_prompt = self._build_prompt(question, schema_info)

        # 步骤3：调用 LLM 生成 SQL
        raw_sql = self.llm.generate(enhanced_prompt)

        # 步骤4：提取 SQL（处理可能的额外文本）
        sql = self._extract_sql(raw_sql)

        # 步骤5：SQL 安全检查
        safety = self.safety_checker.check(sql, schema_info.schema)

        if safety.level == SQLSafetyLevel.FORBIDDEN:
            self._log_audit(user_id, question, sql, "BLOCKED", safety.issues)
            return {
                "success": False,
                "reason": "SQL 未通过安全检查",
                "issues": safety.issues
            }

        # 步骤6：验证 SQL 语法（EXPLAIN 而非实际执行）
        syntax_ok = self._validate_syntax(sql)
        if not syntax_ok:
            return {
                "success": False,
                "reason": "SQL 语法验证失败",
                "raw_sql": sql
            }

        # 步骤7：缓存并返回
        result = {
            "success": True,
            "sql": sql,
            "safety_level": safety.level.value,
            "safety_issues": safety.issues,
            "safety_suggestions": safety.suggestions,
            "schema_used": schema_info.tables_used,
            "generation_time_ms": int((time.time() - start_time) * 1000)
        }

        self.cache[cache_key] = result
        self._log_audit(user_id, question, sql, "SUCCESS", safety.issues)

        return result

    def _build_prompt(self, question: str, schema_info) -> str:
        """构建增强 Prompt"""
        return f"""你是一个 SQL 专家。请根据以下信息生成准确的 SQL 查询。

## 相关表结构
{schema_info.relevant_ddl}

## 业务术语定义
{schema_info.business_glossary}

## 字段说明
{schema_info.column_descriptions}

## 用户问题
{question}

## 要求
1. 只生成 SELECT 语句
2. 优先使用 EXISTS 而非 IN
3. 大表查询必须带 WHERE 和时间范围
4. 金额计算使用 ROUND(..., 2) 保留两位小数
5. 只返回 SQL 代码，不要任何解释

## SQL
"""

    def _extract_sql(self, raw_output: str) -> str:
        """从 LLM 输出中提取 SQL"""
        # 处理 markdown 代码块
        patterns = [
            r'```sql\s*(.*?)```',
            r'```\s*(.*?)```',
            r'(SELECT\s+.*?)(?:\n\s*--|\Z)',
        ]
        for pattern in patterns:
            match = re.search(pattern, raw_output, re.DOTALL | re.IGNORECASE)
            if match:
                return match.group(1).strip()
        return raw_output.strip()

    def _validate_syntax(self, sql: str) -> bool:
        """通过 EXPLAIN 验证 SQL 语法"""
        try:
            # 使用 EXPLAIN 而非实际执行
            self.db.run(f"EXPLAIN {sql}")
            return True
        except Exception:
            return False

    def _log_audit(self, user_id, question, sql, status, issues):
        """记录审计日志"""
        self.audit_log.append({
            "timestamp": time.time(),
            "user_id": user_id,
            "question": question,
            "sql": sql,
            "status": status,
            "issues": issues
        })


# 使用示例
service = TextToSQLService(
    db_uri="postgresql://user:pass@localhost/ecommerce",
    llm_client=your_llm_client,
    schema_store=your_schema_store
)

result = service.generate_sql(
    question="今年Q1华东区域退货率最高的前5个城市",
    user_id="user_001"
)

if result["success"]:
    print(f"生成的 SQL：\n{result['sql']}")
    print(f"安全等级：{result['safety_level']}")
    if result['safety_issues']:
        print(f"⚠️ 安全提示：{result['safety_issues']}")
else:
    print(f"生成失败：{result['reason']}")

四、提升 Text-to-SQL 准确率的 8 个实战技巧

技巧1：高质量 DDL 注入

这是提升准确率最有效的方法，没有之一。

# ❌ 差的 DDL（只有表名和字段类型）
CREATE TABLE orders (id INT, customer_id INT, amount DECIMAL, date DATE);

# ✅ 好的 DDL（含注释和约束语义）
CREATE TABLE orders (
    id BIGINT PRIMARY KEY COMMENT '订单ID，雪花算法生成',
    customer_id BIGINT NOT NULL COMMENT '客户ID，关联customers.id',
    order_date DATE NOT NULL COMMENT '下单日期',
    total_amount DECIMAL(12,2) NOT NULL COMMENT '订单总金额，含税',
    discount_amount DECIMAL(12,2) DEFAULT 0 COMMENT '优惠金额',
    pay_amount DECIMAL(12,2) GENERATED ALWAYS AS (total_amount - discount_amount) STORED COMMENT '实付金额',
    status VARCHAR(20) NOT NULL COMMENT '订单状态: pending/paid/shipped/completed/cancelled/refunded',
    region VARCHAR(50) COMMENT '大区: 华东/华南/华北/华中/西南/西北/东北',
    city VARCHAR(50) COMMENT '城市',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_customer_date (customer_id, order_date),
    INDEX idx_region_date (region, order_date),
    INDEX idx_status (status)
) COMMENT '订单主表，存储所有订单信息';

字段注释是 AI 理解业务语义的关键线索。amount 和 total_amount 含税 这两种描述，对 AI 来说完全不同。

技巧2：业务术语词典

很多业务术语在表结构中找不到直接对应的字段：

BUSINESS_GLOSSARY = """
术语定义：
- 营收 = 订单实付金额之和（不含退款）
- 退货率 = 退货订单数 / 总订单数 × 100%
- 客单价 = 总营收 / 有消费的客户数
- 复购率 = 有2次以上购买记录的客户数 / 总客户数 × 100%
- VIP客户 = vip_level >= 2 的客户
- 沉默客户 = 最近90天无购买记录的客户
- 活跃客户 = 最近30天有购买记录的客户
- 大区划分：
  - 华东 = city IN ('上海','南京','杭州','苏州','宁波','合肥')
  - 华南 = city IN ('广州','深圳','东莞','佛山','厦门')
  - 华北 = city IN ('北京','天津','石家庄','太原')
- Q1 = 1月至3月，Q2 = 4月至6月，Q3 = 7月至9月，Q4 = 10月至12月
"""

技巧3：Few-shot 示例训练

提供 20-50 个高质量的问题-SQL 对作为参考示例：

training_examples = [
    {
        "question": "上个月的总销售额",
        "sql": "SELECT ROUND(SUM(pay_amount), 2) AS total_sales FROM orders WHERE status != 'cancelled' AND order_date >= DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month') AND order_date < DATE_TRUNC('month', CURRENT_DATE)"
    },
    {
        "question": "各城市的订单数量和总金额，按金额降序排列",
        "sql": "SELECT city, COUNT(*) AS order_count, ROUND(SUM(pay_amount), 2) AS total_amount FROM orders WHERE status != 'cancelled' AND order_date >= DATE_TRUNC('month', CURRENT_DATE) GROUP BY city ORDER BY total_amount DESC LIMIT 20"
    },
    {
        "question": "退货率最高的10个城市",
        "sql": """
        WITH order_stats AS (
            SELECT city,
                   COUNT(*) AS total_orders,
                   SUM(CASE WHEN status = 'refunded' THEN 1 ELSE 0 END) AS refund_orders
            FROM orders
            WHERE order_date >= DATE_TRUNC('year', CURRENT_DATE)
            GROUP BY city
        )
        SELECT city,
               total_orders,
               refund_orders,
               ROUND(refund_orders * 100.0 / NULLIF(total_orders, 0), 2) AS refund_rate
        FROM order_stats
        WHERE total_orders > 10
        ORDER BY refund_rate DESC
        LIMIT 10
        """
    },
    {
        "question": "今年每个月的新增客户数量趋势",
        "sql": "SELECT DATE_FORMAT(register_date, '%Y-%m') AS month, COUNT(*) AS new_customers FROM customers WHERE register_date >= DATE_TRUNC('year', CURRENT_DATE) GROUP BY DATE_FORMAT(register_date, '%Y-%m') ORDER BY month"
    },
    {
        "question": "客单价同比去年增长了百分之多少",
        "sql": """
        WITH this_year AS (
            SELECT ROUND(SUM(pay_amount) * 1.0 / COUNT(DISTINCT customer_id), 2) AS avg_order_value
            FROM orders
            WHERE status != 'cancelled'
              AND order_date >= DATE_TRUNC('year', CURRENT_DATE)
        ),
        last_year AS (
            SELECT ROUND(SUM(pay_amount) * 1.0 / COUNT(DISTINCT customer_id), 2) AS avg_order_value
            FROM orders
            WHERE status != 'cancelled'
              AND order_date >= DATE_TRUNC('year', CURRENT_DATE - INTERVAL '1 year')
              AND order_date < DATE_TRUNC('year', CURRENT_DATE)
        )
        SELECT
            this_year.avg_order_value AS this_year_aov,
            last_year.avg_order_value AS last_year_aov,
            ROUND((this_year.avg_order_value - last_year.avg_order_value) * 100.0 / NULLIF(last_year.avg_order_value, 0), 2) AS yoy_growth_percent
        FROM this_year, last_year
        """
    },
]

示例质量 > 数量。 5 个精心编写的示例比 50 个粗糙的示例更有价值。

技巧4：分步生成（Decomposition）

复杂查询不要让模型一次性生成完整 SQL，而是分步走：

async def generate_sql_step_by_step(question: str, llm, schema):
    # 第一步：意图分解
    decomposition = await llm.generate(f"""
    分析以下数据查询需求，将其分解为多个子查询步骤：
    问题：{question}
    表结构：{schema}
    
    格式：
    步骤1: [描述]
    步骤2: [描述]
    ...
    """)

    # 第二步：逐个生成子查询
    sql_parts = []
    for step in decomposition.steps:
        sub_sql = await llm.generate(f"""
        根据以下信息生成 SQL：
        当前步骤：{step.description}
        已生成的 SQL：{chr(10).join(sql_parts)}
        表结构：{schema}
        """)
        sql_parts.append(sub_sql)

    # 第三步：合并为最终 SQL
    final_sql = await llm.generate(f"""
    将以下子查询合并为一个优化的 SQL（使用 CTE）：
    {chr(10).join(sql_parts)}
    表结构：{schema}
    要求：使用 WITH 子句组织，确保可读性和性能
    """)
    return final_sql

技巧5：SQL 验证和自修正循环

def validate_and_fix(sql: str, db, llm, max_retries: int = 3):
    """执行-验证-修正循环"""
    for attempt in range(max_retries):
        try:
            # 在只读副本上执行 EXPLAIN
            plan = db.execute(f"EXPLAIN ANALYZE {sql}")
            cost = parse_execution_cost(plan)

            if cost > THRESHOLD_HIGH:
                # 成本过高，让 LLM 优化
                optimized = llm.generate(f"""
                以下 SQL 的执行计划显示成本过高（{cost}）：
                {sql}
                执行计划：{plan}
                
                请优化此 SQL，主要考虑：
                1. 添加合适的索引提示
                2. 避免 SELECT *
                3. 使用 JOIN 替代子查询
                4. 添加更精确的 WHERE 条件
                """)
                sql = extract_sql(optimized)
                continue

            return {"sql": sql, "cost": cost, "attempts": attempt + 1}

        except db.Error as e:
            error_msg = str(e)
            if attempt < max_retries - 1:
                fixed = llm.generate(f"""
                以下 SQL 执行报错：
                {sql}
                错误信息：{error_msg}
                表结构：{schema}
                请修正错误后返回新的 SQL。
                """)
                sql = extract_sql(fixed)
            else:
                return {"error": error_msg, "sql": sql}

技巧6：结果解释——让 SQL 变成人话

def explain_result(sql: str, result, question, llm):
    """用自然语言解释查询结果"""
    explanation = llm.generate(f"""
    用户问题：{question}
    执行的 SQL：{sql}
    查询结果：
    {format_as_table(result)}
    
    请用简洁的中文解释这些数据，包括关键发现和洞察。
    不要重复数据，只说结论。
    """)
    return explanation

技巧7：缓存策略——相同问题不重复生成

from collections import OrderedDict
import hashlib
import time

class QueryCache:
    """LRU 缓存，避免重复生成相同的 SQL"""

    def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl_seconds

    def get(self, question: str, schema_hash: str) -> Optional[str]:
        key = f"{hashlib.md5(question.encode()).hexdigest()}:{schema_hash}"
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.cache.move_to_end(key)
                return entry["sql"]
            else:
                del self.cache[key]
        return None

    def put(self, question: str, schema_hash: str, sql: str):
        key = f"{hashlib.md5(question.encode()).hexdigest()}:{schema_hash}"
        self.cache[key] = {"sql": sql, "timestamp": time.time()}
        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)

技巧8：用户反馈闭环

class FeedbackLoop:
    """收集用户反馈，持续改进 SQL 生成质量"""

    def collect_feedback(self, user_id: str, question: str,
                        generated_sql: str, was_correct: bool,
                        corrected_sql: Optional[str] = None):
        """记录反馈并更新训练数据"""
        entry = {
            "user_id": user_id,
            "question": question,
            "generated_sql": generated_sql,
            "was_correct": was_correct,
            "corrected_sql": corrected_sql,
            "timestamp": time.time()
        }

        if was_correct:
            # 正确的样本加入训练集
            self.training_data.add_positive(question, generated_sql)
        elif corrected_sql:
            # 修正的样本也加入训练集（覆盖原来的）
            self.training_data.add_corrected(question, corrected_sql)

    def get_daily_report(self):
        """生成每日准确率报告"""
        total = len(self.feedback_today)
        correct = sum(1 for f in self.feedback_today if f["was_correct"])
        return {
            "date": date.today().isoformat(),
            "total_queries": total,
            "correct_queries": correct,
            "accuracy": correct / total if total > 0 else 0,
            "top_error_patterns": self._analyze_error_patterns()
        }

五、性能优化与成本控制

5.1 Schema 索引策略

数据库有 200 张表时，把所有 DDL 塞进 Prompt 不现实。需要智能选择相关表：

import numpy as np
from sentence_transformers import SentenceTransformer

class SchemaRetriever:
    """基于语义相似度的表结构检索器"""

    def __init__(self, schema_data: list[dict]):
        # schema_data: [{"table": "orders", "ddl": "...", "description": "..."}]
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.schema_data = schema_data

        # 预计算所有表的向量
        self.table_embeddings = []
        for item in schema_data:
            text = f"{item['table']} {item['description']} {item['ddl']}"
            embedding = self.model.encode(text)
            self.table_embeddings.append(embedding)
        self.table_embeddings = np.array(self.table_embeddings)

    def retrieve(self, question: str, top_k: int = 5) -> list[dict]:
        """检索与问题最相关的表"""
        question_embedding = self.model.encode(question)

        # 计算余弦相似度
        similarities = np.dot(self.table_embeddings, question_embedding) / (
            np.linalg.norm(self.table_embeddings, axis=1) *
            np.linalg.norm(question_embedding)
        )

        # 获取 top-k
        top_indices = np.argsort(similarities)[-top_k:][::-1]

        return [
            {
                "table": self.schema_data[i]["table"],
                "ddl": self.schema_data[i]["ddl"],
                "relevance_score": float(similarities[i])
            }
            for i in top_indices
        ]

5.2 LLM 成本优化

class CostOptimizedLLM:
    """分级 LLM 调用策略：简单查询用小模型，复杂查询用大模型"""

    def __init__(self):
        self.small_model = "gemini-2.0-flash"      # 便宜、快
        self.medium_model = "gemini-2.5-pro"        # 平衡
        self.large_model = "gemini-3.1-pro"         # 最强、贵

    def classify_complexity(self, question: str, schema_info: dict) -> str:
        """评估查询复杂度"""
        indicators = {
            "simple": 0,    # 单表、简单条件
            "medium": 0,     # 多表 JOIN、基本聚合
            "complex": 0     # 窗口函数、CTE、多步计算
        }

        # 检测多表需求
        if any(word in question for word in ['各', '每', '分', '按', '分别']):
            indicators["medium"] += 2

        # 检测时间比较
        if any(word in question for word in ['同比', '环比', '增长', '变化', '趋势']):
            indicators["complex"] += 3

        # 检测多步计算
        if any(word in question for word in ['率', '占比', '比例', 'TOP', '排名']):
            indicators["complex"] += 2

        # 检测简单查询
        if any(word in question for word in ['多少', '几个', '总数', '总']):
            indicators["simple"] += 2

        max_level = max(indicators, key=indicators.get)
        if indicators[max_level] == 0:
            return "simple"
        return max_level

    def generate(self, question: str, schema_info: dict):
        """根据复杂度选择模型"""
        complexity = self.classify_complexity(question, schema_info)

        model_map = {
            "simple": self.small_model,
            "medium": self.medium_model,
            "complex": self.large_model
        }

        selected_model = model_map[complexity]
        # 调用对应模型...
        return selected_model

成本对比（估算）：

对比全用大模型（$15/天），节省约 60%。

5.3 结果缓存与预计算

from functools import lru_cache
from datetime import datetime, timedelta
import json

class ResultCache:
    """智能结果缓存：时间相关查询按时效性缓存"""

    def __init__(self, redis_client, default_ttl=300):
        self.redis = redis_client
        self.default_ttl = default_ttl

    def _get_ttl(self, sql: str) -> int:
        """根据 SQL 时间粒度决定缓存时间"""
        if 'CURRENT_DATE' in sql or 'NOW()' in sql:
            if 'HOUR' in sql or 'MINUTE' in sql:
                return 60      # 分钟级数据缓存1分钟
            elif 'DATE_TRUNC' in sql or 'month' in sql.lower():
                return 3600    # 日级数据缓存1小时
            else:
                return 300     # 默认5分钟
        return 86400  # 历史数据缓存24小时

    async def get_or_compute(self, sql: str, compute_fn):
        cache_key = f"sql_result:{hashlib.md5(sql.encode()).hexdigest()}"

        # 尝试从缓存获取
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # 计算结果
        result = await compute_fn(sql)

        # 写入缓存
        ttl = self._get_ttl(sql)
        await self.redis.setex(cache_key, ttl, json.dumps(result))

        return result

六、真实落地案例：电商数据自助查询平台

6.1 系统架构

┌──────────────────────────────────────────────┐
│                  前端 (React)                  │
│  ┌─────────────────────────────────────────┐ │
│  │  自然语言输入框                            │ │
│  │  "今年Q1华东区退货率最高的前10个城市"       │ │
│  └─────────────────────────────────────────┘ │
│  ┌─────────────────────────────────────────┐ │
│  │  生成的 SQL（可编辑）                      │ │
│  │  结果表格 + 图表可视化                    │ │
│  │  结果解释（自然语言）                      │ │
│  │  👍👎 反馈按钮                            │ │
│  └─────────────────────────────────────────┘ │
└──────────────────┬───────────────────────────┘
                   │ REST API
┌──────────────────▼───────────────────────────┐
│              后端服务 (FastAPI)                │
│  ┌──────────────┐  ┌──────────────────────┐  │
│  │ Query Router   │  │ Schema Retriever    │  │
│  │ (复杂度分流)    │  │ (语义检索相关表)     │  │
│  └──────┬───────┘  └──────────┬───────────┘  │
│         ▼                     ▼              │
│  ┌──────────────────────────────────────┐    │
│  │  SQL Generation Pipeline              │    │
│  │  1. Schema Linking (表结构映射)        │    │
│  │  2. Prompt Assembly (提示词组装)        │    │
│  │  3. LLM Call (模型调用)               │    │
│  │  4. SQL Extraction (SQL提取)          │    │
│  │  5. Safety Check (安全检查)            │    │
│  │  6. Validation (语法验证)             │    │
│  │  7. Self-Correction (自修正)           │    │
│  └──────────────────────────────────────┘    │
│  ┌──────────────┐  ┌──────────────────────┐  │
│  │ Query Cache    │  │ Audit Logger        │  │
│  │ (Redis)        │  │ (审计日志)          │  │
│  └──────────────┘  └──────────────────────┘  │
└──────────────────┬───────────────────────────┘
                   │ 只读连接
┌──────────────────▼───────────────────────────┐
│         数据库 (PostgreSQL 只读副本)            │
│  orders, order_items, customers, products,    │
│  returns, inventory, promotions, ...          │
└───────────────────────────────────────────────┘

6.2 API 接口设计

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
import asyncio

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


class QueryRequest(BaseModel):
    question: str
    user_id: str
    session_id: Optional[str] = None  # 多轮对话上下文


class QueryResponse(BaseModel):
    success: bool
    sql: Optional[str] = None
    result: Optional[list[dict]] = None
    explanation: Optional[str] = None
    safety_level: Optional[str] = None
    safety_warnings: Optional[list[str]] = None
    generation_time_ms: int = 0
    cached: bool = False


@app.post("/api/query", response_model=QueryResponse)
async def text_to_sql(request: QueryRequest):
    """
    自然语言查询接口

    将用户的自然语言问题转换为 SQL 并执行
    """
    start = time.time()

    # 权限检查
    user = await get_user(request.user_id)
    if not user.has_permission("data_query"):
        raise HTTPException(403, "无数据查询权限")

    # 限流检查
    if await is_rate_limited(request.user_id):
        raise HTTPException(429, "查询过于频繁，请稍后再试")

    # 生成 SQL
    result = await service.generate_sql(
        question=request.question,
        user_id=request.user_id
    )

    if not result["success"]:
        return QueryResponse(
            success=False,
            explanation=result.get("reason", "生成失败")
        )

    # 执行查询（只读副本）
    try:
        data = await readonly_db.execute(result["sql"])
        rows = [dict(row) for row in data]
    except Exception as e:
        return QueryResponse(
            success=False,
            sql=result["sql"],
            explanation=f"SQL 执行失败：{str(e)}"
        )

    # 生成自然语言解释
    explanation = await service.explain_result(
        question=request.question,
        sql=result["sql"],
        result=rows
    )

    return QueryResponse(
        success=True,
        sql=result["sql"],
        result=rows,
        explanation=explanation,
        safety_level=result.get("safety_level"),
        safety_warnings=result.get("safety_issues"),
        generation_time_ms=int((time.time() - start) * 1000),
        cached=result.get("cached", False)
    )


@app.post("/api/feedback")
async def submit_feedback(
    query_id: str,
    correct: bool,
    corrected_sql: Optional[str] = None
):
    """提交 SQL 质量反馈"""
    await service.collect_feedback(query_id, correct, corrected_sql)
    return {"status": "ok"}

6.3 实际效果数据

根据已部署 Text-to-SQL 系统的企业反馈：

七、Text-to-SQL 的局限性与未来展望

7.1 当前局限

1. 准确率天花板

即使 Gemini-SQL2 达到了 80.04% 的 BIRD 准确率，这意味着每 5 个复杂查询中就有 1 个是错的。在生产环境中，这个错误率需要通过人工复核来兜底。

2. Schema 演化问题

当数据库表结构发生变化（加字段、改名、拆表）时，Text-to-SQL 系统需要同步更新训练数据。这个「维护成本」往往被低估。

3. 领域知识依赖

BIRD 评测需要外部知识支持。真实业务中，「高价值客户」的定义可能每个月都在变——AI 的知识需要持续更新。

4. 性能不可控

AI 生成的 SQL 可能执行很快，也可能跑满整个数据库。虽然我们有了安全检查层，但很难预测所有性能风险。

7.2 技术趋势

趋势一：Agentic Text-to-SQL

不再是「一个问题生成一条 SQL」，而是：

用户问题 → Agent 拆解任务 → 逐个查询 → 中间结果分析 → 组合回答

代表项目：APEX-SQL、AgentTrek

趋势二：多模态数据库理解

未来的 Text-to-SQL 系统不仅能读取 DDL，还能：

• 分析查询日志，学习常用查询模式
• 理解 ETL 流程，知道数据的来龙去脉
• 读取数据字典和 BI 报表定义

趋势三：从「生成 SQL」到「生成答案」

终极形态不是让 AI 写 SQL，而是：

用户问：为什么上个月华东区退货率突然升高？
AI：经过分析，退货率从 3.2% 升至 5.1%。主要原因是：
    1. SKU-1234 的质量问题导致该单品退货量增加 340%
    2. 促销活动带来的冲动消费退货增加
    3. 建议：下架 SKU-1234 并调整促销策略

AI 不仅查数据，还分析原因、给出建议。这已经是 Agent BI 的范畴了。

7.3 给开发者的建议

1. 从 Vanna 开始，不要从零造轮子——它的 RAG 架构已经解决了最核心的 Schema Linking 问题
2. 安全第一：永远在只读副本上执行 AI 生成的 SQL，永远加安全检查层
3. 渐进式上线：先给内部数据团队用，收集反馈改进，再开放给业务人员
4. 建立训练数据飞轮：每次人工修正的 SQL 都要回流到训练集
5. 监控准确率：建立日常准确率看板，准确率低于 80% 就要介入调优
6. 设定预期：告诉用户这是「辅助工具」而非「自动化工具」，复杂查询仍需人工复核

八、总结

Gemini-SQL2 的 80.04% BIRD 准确率是 Text-to-SQL 领域的一个重要里程碑。它证明了基于大模型的专用微调可以显著超越通用模型的 few-shot 能力。

但从工程视角看，Text-to-SQL 要真正在企业落地，需要的不只是更准确的模型，而是一套完整的工程体系：

• 数据层：高质量的 DDL 注入、业务术语词典、Few-shot 示例训练
• 模型层：分级模型调用（简单用小模型，复杂用大模型）
• 安全层：SQL 安全检查、只读执行、行级权限控制
• 验证层：语法验证、执行计划分析、自修正循环
• 运维层：结果缓存、审计日志、准确率监控、反馈闭环

Text-to-SQL 的终局不是让 AI 写 SQL，而是让「人人都是数据分析师」从口号变成现实。 Gemini-SQL2 让我们离这个目标又近了一步，但最后的 20% 路程——安全、治理、准确性——仍然需要工程师去填。

如果你正在考虑在项目中引入 Text-to-SQL，现在就是最好的时机。先从 Vanna + 你的数据库开始，用 5 分钟搭一个原型，用 5 周打磨生产级系统。你会发现，这个技术已经从「学术玩具」变成了「工程利器」。

参考资源：

• BIRD Benchmark - Text-to-SQL 评测基准
• Vanna AI - 开源 Text-to-SQL RAG 框架
• Gemini-SQL2 发布公告 - Google Research 官方发布
• PremSQL - 本地 Text-to-SQL 方案
• LangChain SQL Agent - LangChain SQL 集成