Text2SQL 学习笔记：从「直接生成 SQL」到可纠错、可规划的数据库智能体

写作主线与《Agentic RAG》笔记一致：以演进逻辑贯穿——每一代方案都在补上一代的具体短板。每个知识点讲清 是什么（机制）→ 为什么这么设计（动机）→ 怎么实现（最小可运行代码）。代码栈：Python + LangChain/LangGraph + 内置 SQLite（零数据库部署），配套可逐格运行的 text2sql_demo.ipynb，已用 DeepSeek deepseek-v4-flash 实测通过。

学习笔记的配套代码： https://github.com/LT-IENG/AI-Agent-Study-Notes

0. 导读
1. 背景：Text2SQL 是什么、为什么重要
2. 完整链路与六大难点
3. Naive Text2SQL：把 Schema 全塞进去直接生成
4. 关键技术一：Schema Linking（模式链接，最核心）
5. 关键技术二：上下文增强（元数据 / Few-shot / 方言 / 枚举值）
6. 关键技术三：执行反馈自纠错闭环
7. 经典学术方案速览：DIN-SQL / DAIL-SQL / MAC-SQL / CHESS
8. Agentic Text2SQL：把数据库能力变成工具
9. 评测体系：EX / EM / VES 与 Spider / BIRD
10. 生产化：安全、歧义澄清、结果解释与成本
11. 方案对比与选型决策
12. 参考资料

0. 导读

一句话抓住主线：

Text2SQL 的进化史，就是不断和「模型会一本正经地编造一个跑不通（或跑通但算错）的 SQL」这件事作斗争的历史。

Naive Text2SQL：把全部表结构 + 问题丢给 LLM，一把生成 SQL。简单，但库一大就崩、还会编造列名。
Schema Linking：先缩小范围，只把「和问题相关的表和列」给模型——这是决定上限的一步。
上下文增强：用列注释、外键关系、枚举值、相似问题的标准 SQL 范例（Few-shot）补齐「业务语义」。
执行反馈自纠错：SQL 不是写完就完，先真的跑一次，报错/空结果就把反馈喂回让模型改，形成闭环。
Agentic Text2SQL：把「看表清单、看表结构、看样例数据、执行 SQL、查指标口径」做成工具，让模型像分析师一样多轮探索、自我纠错、拆解复杂问题。

学习建议：第 3 章的缺陷清单是后面所有章节的「问题来源」，请先吃透；第 4、6、8 章是工程上 ROI 最高的三块。

1. 背景：Text2SQL 是什么、为什么重要

1.1 是什么

Text2SQL（也叫 NL2SQL，Natural Language to SQL）：把人类的自然语言问题（「上个季度华东区销售额 Top10 的商品是哪些？」）自动翻译成可在数据库执行的 SQL 查询，并（通常）执行后把结果/图表返回给用户。

flowchart LR
    U[自然语言问题] --> S[Text2SQL 系统]
    S --> SQL[结构化 SQL]
    SQL --> DB[(数据库)]
    DB --> R[结果集]
    R --> N[自然语言解读/可视化]
    N --> U

1.2 为什么重要：数据分析的「最后一公里」民主化

企业里 90% 的数据躺在数据库里，但能写 SQL 的人是少数。传统 BI 报表是「预先做好的固定问题」，而业务的问题是临时、开放、变化的。Text2SQL 让任何人用一句话直接查询数据，是「自助式数据分析（Self-service Analytics）」和企业数据 Copilot 的核心底座，也是通用 Agent 操作结构化数据时必备的工具能力。

1.3 本质：它是「结构化数据版的 RAG」

把 Text2SQL 和你已经熟悉的 RAG 对照，会非常好理解：

对比维度	向量 RAG	Text2SQL
数据形态	非结构化文本（文档）	结构化数据（表、行、列）
「检索」方式	向量相似度找文本块	生成 SQL 精确查询/聚合
知识载体	切块后的文档	Schema（表/列/关系）+ 数据本身
可计算性	弱（只能引用原文）	强（SUM/JOIN/GROUP BY/窗口函数精确计算）
主要风险	检索不到、张冠李戴	编造表列名、SQL 报错、逻辑算错
验证手段	人看引用是否对	直接执行，结果可判定对错

关键洞察：Text2SQL 拥有 RAG 没有的「可执行验证」优势——SQL 对不对，跑一下数据库、比对结果就知道，这使得「执行反馈闭环」成为可能（第 6 章）。但它也更脆弱：SQL 有严格语法和精确的表/列名，一个标识符错了就彻底失败。

1.4 典型应用形态

BI 问答助手：对接数仓（ClickHouse/Hive/Snowflake/MySQL），自然语言出数、出图。
数据 Agent 的工具：通用 Agent 在完成任务时，把 Text2SQL 当作查询数据库的一个 Tool。
数据治理/运维：自然语言查表、查指标、生成报表 SQL 草稿。
开源产品形态：Vanna、WrenAI、Dataherald、SuperSQL，以及 LangChain 内置的 SQL Agent。

2. 完整链路与六大难点

2.1 一次完整 Text2SQL 的链路

flowchart TD
    Q[用户自然语言问题] --> CL[意图识别/澄清<br/>是否需要查库?有无歧义?]
    CL --> SL["① Schema Linking<br/>选出相关表/列/关系/枚举值"]
    SL --> CTX["② 上下文组装<br/>Schema + 注释 + 外键 + Few-shot + 方言"]
    CTX --> GEN["③ SQL 生成"]
    GEN --> VAL["④ 校验/修正<br/>语法检查 · 标识符核对 · 安全检查"]
    VAL --> EXE["⑤ 执行（dry-run / 限量执行）"]
    EXE --> OK{执行成功且结果合理?}
    OK -->|报错/空结果/异常| FIX["⑥ 反馈纠错：把错误喂回重写"] --> GEN
    OK -->|是| ANS[⑦ 结果解读 + 可视化 + 引用]

2.2 六大难点（也是后续技术的「问题清单」）

Schema 规模问题（最现实的工程难点）：真实数仓常有成百上千张表、上万列。不可能把全部 DDL 塞进 Prompt（超 token、且模型在巨大 schema 里找不准）。→ 第 4 章 Schema Linking。
语义鸿沟 / 业务术语歧义：用户说「活跃用户」「GMV」「华东」，数据库里可能是 is_active=1、pay_amt、region_code IN (...)。业务口径不在 SQL 里，在人的脑子里。→ 第 5 章元数据/指标口径增强。
列名/表名幻觉：模型会「想当然」编一个不存在的列（如把 user_name 写成 username），SQL 直接报错。
复杂结构：多表 JOIN、嵌套子查询、GROUP BY/HAVING、窗口函数、日期处理、多跳关系；问题越复杂，一次生成越容易错。
多方言差异：MySQL / PostgreSQL / Hive / ClickHouse / SQLite 函数与语法不同（日期函数、分页、类型转换尤其不同），不声明方言就会生成「在目标库跑不通」的 SQL。
「跑通了但算错」：SQL 语法正确、能执行，但 JOIN 漏了条件导致笛卡尔积、聚合粒度错、NULL 处理错、时间窗口错——语法正确 ≠ 语义正确，这是最难的一类。

记住：难点 1、2、3 靠「给对上下文」解决；难点 4 靠「分解 + 工具化」解决；难点 6 靠「执行反馈 + 结果校验 + 业务口径」兜住。

3. Naive Text2SQL：把 Schema 全塞进去直接生成

3.1 是什么

最朴素的做法：用程序读出数据库全部建表语句（DDL），连同问题一起塞进 Prompt，让 LLM 直接输出 SQL。

flowchart LR
    DDL[全部表 DDL] --> P[Prompt]
    Q[问题] --> P
    P --> LLM[LLM] --> SQL[SQL]

3.2 怎么实现（最小可运行，SQLite + LangChain）

完整可运行版见 text2sql_demo.ipynb。LangChain 的 SQLDatabase 封装了「连接库、抽取 DDL、执行 SQL、做方言转义」等脏活，是理解后续一切的基础。

① 先造一个迷你电商库（零部署，SQLite 内置）

# build_demo_db.py —— 生成演示数据库（Notebook 会自动执行）
import sqlite3
conn = sqlite3.connect("demo_ecommerce.db")
cur = conn.cursor()
cur.executescript("""
CREATE TABLE users(
    user_id INTEGER PRIMARY KEY,
    name TEXT,
    region TEXT,          -- 华东/华北/华南...
    is_active INTEGER     -- 1=活跃
);
CREATE TABLE products(
    product_id INTEGER PRIMARY KEY,
    product_name TEXT,
    category TEXT,
    price REAL
);
CREATE TABLE orders(
    order_id INTEGER PRIMARY KEY,
    user_id INTEGER,
    product_id INTEGER,
    amount REAL,
    order_time TEXT,      -- ISO 时间字符串
    FOREIGN KEY(user_id) REFERENCES users(user_id),
    FOREIGN KEY(product_id) REFERENCES products(product_id)
);
""")
# 插入若干演示数据（Notebook 中给出完整 INSERT）
conn.commit(); conn.close()

② 用 SQLDatabase 连接并查看 schema

from langchain_community.utilities.sql_database import SQLDatabase
db = SQLDatabase.from_uri("sqlite:///demo_ecommerce.db")

print(db.dialect)                 # 方言：sqlite —— 必须告诉 LLM，避免生成错方言
print(db.get_usable_table_names())# 可用表清单
print(db.get_table_info())        # 全部 DDL（Naive 做法会把它整体塞进 Prompt）

③ Naive 生成

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import os

llm = ChatOpenAI(
    model=os.getenv("CHAT_MODEL", "deepseek-v4-flash"),
    base_url=os.getenv("CHAT_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.getenv("LLM_API_KEY"),
    temperature=0,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 {dialect} 数据库专家，只能输出可执行 SQL，不要解释，不要 markdown 代码块。"),
    ("human", "数据库 Schema：\n{schema}\n\n问题：{question}\nSQL："),
])

chain = prompt | llm
sql = chain.invoke({
    "dialect": db.dialect,
    "schema": db.get_table_info(),       # Naive：全量 schema
    "question": "华东区每个品类的总销售额是多少？按金额降序。",
}).content.strip()
print(sql)

④ 执行（run 自带方言转义与只读保护的雏形）

result = db.run(sql)   # 内部用 SQLAlchemy 执行，返回字符串化的结果
print(result)

3.3 Naive 的四个致命缺陷

Schema 不可扩展：表一多，get_table_info() 直接撑爆上下文，且无关表会严重干扰模型（难点 1）。
没有业务语义：只有裸 DDL，模型不知道「活跃用户 = is_active=1」「GMV 只算已支付」（难点 2）。
一锤子买卖，错了没人管：生成的 SQL 报错或算错，没有任何纠正机制（难点 3、6）。
复杂问题无法拆解：多跳、嵌套问题被要求「一步写出最终 SQL」，出错率随复杂度指数上升（难点 4）。

下面三章分别解决「给对上下文（4、5 章）」和「错了能改（6 章）」。

4. 关键技术一：Schema Linking（模式链接，最核心）

4.1 是什么 & 为什么它决定上限

Schema Linking（模式链接 / Schema 定位）：在写 SQL 之前，先判断「这个问题需要用到哪些表、哪些列、哪些外键关系（join path）、甚至哪些具体取值（value linking）」，只把这一小部分相关 schema 交给生成模型。

学术界的共识（DIN-SQL、MAC-SQL、CHESS 等）是：Text2SQL 的错误里很大比例源于 schema 选错，而不是 SQL 语法写错。 这符合直觉——表都选错了，SQL 写得再漂亮也是错的。它同时解决三个问题：省 token、降干扰、减少列名幻觉。

flowchart TD
    Q[问题] --> T[① 表级筛选<br/>几十上百张表 → 相关的几张]
    T --> C[② 列级筛选<br/>每张表只留相关列 + 主键/外键]
    C --> J[③ 连接路径<br/>沿外键确定 JOIN 顺序]
    C --> V[④ 值链接 Value Linking<br/>问题里的“华东”对应 region 的哪些实际取值]
    J --> S[精简后的 Schema 上下文]
    V --> S

4.2 两级筛选：先选表、再选列

表筛选（粗）：表名 + 表注释通常很短，可以把「全部表的名字和一句话注释」交给 LLM（或用向量检索表描述），让它选出相关表，成本很低。
列筛选（细）：只对选中的少数表，把它们的列名 + 列注释交给 LLM 挑相关列；主键和外键列强制保留（否则 JOIN 会断）。

4.3 Value Linking：最容易被忽略的一环

用户说「华东区」，但数据库 region 列里存的可能是 East China、EC 或 {"华东":"EC"} 字典编码。问题中的自然语言值必须先对齐到列里的真实取值，否则 WHERE region='华东' 查出来永远是空。做法：对低基数离散列（地区、状态、品类）预先 SELECT DISTINCT 采样实际取值放进上下文，或维护一张业务术语字典。

4.4 怎么实现（最小可运行）

# schema_linking.py
from langchain_core.prompts import ChatPromptTemplate

# 0) 准备“表目录”：表名 + 注释（真实项目从数据库 COMMENT / 数据字典读取）
TABLE_CATALOG = {
    "users":    "用户表：用户基本信息、所属地区、是否活跃",
    "products": "商品表：商品名称、品类、单价",
    "orders":   "订单表：每笔订单的用户、商品、金额、下单时间",
}
catalog_text = "\n".join(f"- {t}: {d}" for t, d in TABLE_CATALOG.items())

# 1) 表级筛选
table_picker = ChatPromptTemplate.from_template(
    "下面是数据库的表目录：\n{catalog}\n\n"
    "为了回答问题【{q}】，需要哪些表？只输出表名，用逗号分隔，不要解释。"
) | llm
picked_tables = [t.strip() for t in
                 (table_picker.invoke({"catalog": catalog_text, "q": question})
                  .content.split(",")) if t.strip() in TABLE_CATALOG]
print("选中的表：", picked_tables)

# 2) 只取选中表的 DDL（主键/外键天然包含在内），并采样低基数列的真实取值
def build_focused_schema(tables, distinct_cols=None, sample=10):
    ddl = db.get_table_info(table_names=tables)          # 只含相关表，大幅省 token
    extra = []
    for tbl, col in (distinct_cols or []):
        vals = db.run(f"SELECT DISTINCT {col} FROM {tbl} LIMIT {sample}")
        extra.append(f"-- {tbl}.{col} 的实际取值示例：{vals}")  # Value Linking
    return ddl + "\n" + "\n".join(extra)

focused_schema = build_focused_schema(
    picked_tables,
    distinct_cols=[("users", "region")],                 # 对齐“华东”这类自然语言值
)
print(focused_schema)

工程提示：表非常多时，表级筛选用「表描述向量检索 + LLM 复核」比纯 LLM 更省、更稳；列级筛选可以缓存（schema 不变就不用每次重算）。

5. 关键技术二：上下文增强（元数据 / Few-shot / 方言 / 枚举值）

Schema Linking 解决「给哪些表列」，上下文增强解决「怎么让模型读懂这些表列背后的业务」。

5.1 四种增强手段与动机

手段	解决什么	做法
元数据/注释	裸列名 `amt`、`st` 没人看得懂	给每张表、每列加中文业务注释；标注主键外键、单位、口径
业务口径（Metric Layer）	「活跃用户」「GMV」口径不统一	维护指标字典：GMV = 已支付订单金额之和，附标准 SQL 片段
Few-shot 示例	模型不熟悉你的 schema 套路	检索几个「相似问题 → 标准 SQL」范例放进 Prompt
方言 & 约束声明	生成错方言、忘记 LIMIT	明确告知 dialect、只允许 SELECT、默认加 LIMIT、日期函数写法

5.2 Few-shot：动态检索相似「问题→SQL」范例

固定塞几个例子不如「按问题相似度动态挑例子」。注意：Text2SQL 的 few-shot 用问题文本相似度即可，不强依赖向量库——下面用最轻量的方式演示（生产可用 embedding 检索历史正确 SQL）。

# few_shot.py —— 从历史正确案例库中挑最相似的几个作为示范
import difflib

# 历史沉淀下来的“问题 -> 标准SQL”黄金案例（生产中来自审核过的查询日志）
SQL_EXAMPLES = [
    {"q": "各地区订单总额",
     "sql": "SELECT u.region, SUM(o.amount) AS total FROM orders o "
            "JOIN users u ON o.user_id=u.user_id GROUP BY u.region;"},
    {"q": "活跃用户数量",
     "sql": "SELECT COUNT(*) FROM users WHERE is_active=1;"},
]

def pick_examples(question, k=2):
    scored = [(difflib.SequenceMatcher(None, question, e["q"]).ratio(), e)
              for e in SQL_EXAMPLES]
    scored.sort(key=lambda x: x[0], reverse=True)
    return [e for _, e in scored[:k]]

def format_examples(examples):
    return "\n\n".join(f"问题：{e['q']}\nSQL：{e['sql']}" for e in examples)

5.3 把所有增强组装成「生成 Prompt」

# assemble_and_generate.py
GEN_SYS = """你是严格的 {dialect} SQL 专家，遵守：
1. 只能使用给定 Schema 中存在的表和列，禁止编造标识符；
2. 只输出 SELECT 查询，禁止任何增删改操作；默认最多返回 {top_k} 行；
3. 注意业务口径与取值字典；参考给出的 Few-shot 写法；
4. 只输出 SQL 本体，不要解释、不要 markdown 代码块。"""

gen_prompt = ChatPromptTemplate.from_messages([
    ("system", GEN_SYS),
    ("human",
     "【Schema】\n{schema}\n\n【相似范例】\n{examples}\n\n"
     "【问题】{question}\nSQL："),
])

sql_chain = gen_prompt | llm
sql = sql_chain.invoke({
    "dialect": db.dialect,
    "top_k": 100,
    "schema": focused_schema,
    "examples": format_examples(pick_examples(question)),
    "question": question,
}).content.strip()

5.4 为什么这些增强有效（设计原理）

LLM 生成 SQL 本质是「在你给定的约束空间里做模式匹配与组合」。你给的业务信息越精确，它的搜索空间越小、越不容易幻觉。注释和口径压缩了「语义歧义空间」，Few-shot 提供了「输出风格与 JOIN 套路模板」，方言与约束则提前封死了一类低级错误。这些都是一次性建设、可复用、边际成本递减的资产（数据字典、指标库、黄金案例库），也是企业 Text2SQL 真正的护城河。

6. 关键技术三：执行反馈自纠错闭环

6.1 是什么 & 为什么这是 Text2SQL 独有的优势

执行反馈自纠错（Execution-guided Self-Correction）：不假设模型一次写对，而是把「生成 → 执行 → 拿到报错或异常结果 → 把反馈喂回模型修正」做成一个循环，直到成功或达到重试上限。

Text2SQL 比一般文本生成更适合做这件事，因为数据库本身就是一个免费、精确、即时的「评判器」：语法错了数据库会报精确错误（no such column: username），这是最有价值的纠错信号——它直接告诉模型「你编的列名不存在」。

flowchart TD
    G[生成 SQL] --> CHK{静态校验<br/>只允许 SELECT?标识符都存在?}
    CHK -->|不通过| G
    CHK -->|通过| EX[受控执行<br/>LIMIT/只读/超时]
    EX --> R{结果}
    R -->|数据库报错| FB[把错误信息 + 旧SQL 喂回模型修正] --> G
    R -->|空结果| FB2[提示：0行，可能取值/JOIN条件错] --> G
    R -->|正常结果| OUT[输出结果]

6.2 三层校验，层层拦截

静态校验（不连库，最快）：正则/sqlparse 解析，拦截非 SELECT（INSERT/UPDATE/DELETE/DROP/ATTACH/PRAGMA）、补 LIMIT、用 schema 核对表列名是否存在。
执行校验（连库，最准）：在只读事务里 EXPLAIN 或小样本执行，捕获语法/标识符/类型错误。
结果合理性校验：空结果、行数异常多、数值异常大、列全是 NULL——触发「结果反思」。

6.3 怎么实现（LangGraph 版纠错循环）

用 LangGraph 的条件边表达「成功就结束、失败就带着错误回到生成节点」的循环（这也是第 8 章 Agentic 的雏形）：

# self_correct_langgraph.py
import re, sqlparse
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

FORBIDDEN = re.compile(r"\b(insert|update|delete|drop|alter|attach|pragma)\b", re.I)

class State(TypedDict):
    question: str
    schema: str
    sql: str
    error: str
    retries: int
    result: str

def generate(state: State) -> State:
    hint = f"\n上一版 SQL：{state['sql']}\n数据库报错：{state['error']}\n请修正后只输出 SQL。" if state.get("error") else ""
    sql = sql_chain.invoke({
        "dialect": db.dialect, "top_k": 100, "schema": state["schema"],
        "examples": format_examples(pick_examples(state["question"])),
        "question": state["question"] + hint,
    }).content.strip()
    sql = extract_sql(sql)                 # 去掉模型可能加的 markdown 代码块包裹
    return {"sql": sql, "retries": state.get("retries", 0) + 1}

def validate_and_run(state: State) -> State:
    if FORBIDDEN.search(state["sql"]):
        return {"error": "只允许 SELECT 查询"}
    try:
        rows = db.run(state["sql"])        # 受控执行
        if not rows or rows == "[]":
            return {"error": "查询返回 0 行，请检查表名/列名取值与 JOIN 条件"}
        return {"result": rows, "error": ""}
    except Exception as e:
        return {"error": f"{type(e).__name__}: {e}"}   # 精确报错回灌

def route(state: State) -> str:
    if state.get("result"):
        return "done"
    if state["retries"] >= 3:              # 护栏：最多重试 3 次，防止死循环
        return "giveup"
    return "retry"

g = StateGraph(State)
g.add_node("generate", generate)
g.add_node("run", validate_and_run)
g.add_edge(START, "generate")
g.add_edge("generate", "run")
g.add_conditional_edges("run", route, {"done": END, "giveup": END, "retry": "generate"})
corrector = g.compile()

out = corrector.invoke({"question": question, "schema": focused_schema,
                        "sql": "", "error": "", "retries": 0, "result": ""})
print("最终 SQL：", out["sql"]); print("结果：", out.get("result") or out.get("error"))

其中 extract_sql 负责从模型输出里稳妥地抽出 SQL（模型有时会带 markdown 或解释）：

def extract_sql(text: str) -> str:
    FENCE = chr(96) * 3                       # 三个反引号，避免在源码里直接写出
    m = re.search(rf"{FENCE}sql\s*(.*?){FENCE}", text, re.S | re.I)
    return (m.group(1) if m else text).strip().rstrip(";")

6.4 为什么需要「重试上限」和「错误去噪」

必须有最大轮数：模型可能反复在同一个错误上打转，护栏防止 token 烧穿、死循环。
报错要翻译/裁剪：原始数据库错误可能又长又含内部路径，截取关键信息（错误类型 + 附近标识符）效果更好；同类错误重复出现时，应提示模型「换一种思路」而不是重复尝试。
空结果不等于错误，但要反思：很多业务问题的「错」体现在查空，提示模型检查 value linking（第 4.3 节）。

7. 经典学术方案速览：DIN-SQL / DAIL-SQL / MAC-SQL / CHESS

这些论文奠定了今天工程方案的基本套路，理解思想即可，不必逐行复现。

方案	核心思想（为什么这么设计）	关键模块
DIN-SQL（NeurIPS 2023）	把 Text2SQL 分解为有序的 4 步，降低单步难度，曾登顶 Spider	① Schema Linking → ② 问题分类（简单/非嵌套/嵌套）→ ③ 分类生成（复杂问题先解子查询）→ ④ Self-Correction 修小错（漏 DISTINCT/DESC 等）
DAIL-SQL（2023）	证明「提示词工程」就能逼近微调：把问题先让模型假设几个可能 SQL，再据此选 few-shot	高效的 few-shot 选择策略、问题与 SQL 的双重表示
MAC-SQL（2023）	用多智能体协作降低单点错误：Selector 选 schema、Generator 写 SQL、Verifier 审查执行	三 Agent 协作 + 执行验证
CHESS（2024）	面向 BIRD 这类脏数据、大库场景，强化 schema 检索与候选 SQL 融合	分层 schema linking、candidate 生成与排序、执行引导修正

共同方法论（今天依然适用）：先做 schema linking → 按复杂度分解 → 生成 → 用执行结果自我修正。这正是第 4、6 章代码的理论来源；而 MAC-SQL 的「多角色协作」思想，在 Agent 时代演化成了第 8 章的 Agentic 架构。

8. Agentic Text2SQL：把数据库能力变成工具

8.1 是什么：从「固定 Pipeline」到「模型自主探索数据库」

前面的方案无论加多少环节，流程都是开发者写死的。Agentic Text2SQL 则把数据库的各种操作封装成工具（Tools），让 LLM 在 ReAct 循环里像人类数据分析师一样自主探索：先看有哪些表 → 看表结构 → 看几行真实数据搞清楚取值 → 写 SQL → 执行 → 报错就改 → 必要时把大问题拆成多个小查询逐步逼近。

sequenceDiagram
    participant U as 用户
    participant L as LLM 分析师
    participant DB as 数据库工具集
    U->>L: 上个季度华东 Top10 商品？
    L->>DB: list_tables() 看有哪些表
    DB-->>L: users/products/orders
    L->>DB: describe_table(orders, products) 看列
    DB-->>L: DDL + 外键
    L->>DB: sample_rows(users) 看 region 真实取值
    DB-->>L: 华东=East? 实际是“华东”
    L->>DB: run_sql(第一版)
    DB-->>L: 报错 no such column
    L->>DB: run_sql(修正版)
    DB-->>L: 结果集
    L->>U: 答案 + SQL + 简要解读

8.2 一套好用的数据库工具集（由浅入深）

工具	职责	为什么需要（动机）
`list_tables`	列出所有表 + 一句话注释	低成本建立全局认知
`describe_table(names)`	看指定表的 DDL/列注释/外键	替代一次性灌入全库 schema，按需取用
`sample_rows(table, n)`	看几行真实样例数据	解决 value linking：亲眼看到「华东」怎么存的，比猜强一百倍
`run_sql(sql)`	只读、限量、带超时地执行 SQL	唯一的「取证/验证」通道，内置安全护栏
`get_metric_def(name)`	查指标口径字典	解决「GMV/活跃用户」业务定义问题

设计要点：让模型「先探索、后写、写完必跑、错了能改」，并且 run_sql 内部强制第 6.2 节的三层校验。

8.3 怎么实现：LangChain 内置 SQL Agent 与手写版

方式 A：复用官方 SQL 工具集 + 现代 Agent 运行时（生产最快路径）

LangChain 的 SQLDatabaseToolkit 已经把「列表 / 看结构 / 语法检查 / 执行」打磨成了现成工具。注意：老式的 create_sql_agent 基于 legacy AgentExecutor，在推理型模型（如 deepseek-v4-flash，响应带 reasoning_content）的多轮回传中可能报错；LangChain 1.x 的推荐做法是取出官方工具，用现代 Agent 运行时 create_agent 驱动：

# agentic_text2sql_official.py
from langchain.agents import create_agent                      # 1.x 推荐，旧名 create_react_agent
from langchain_community.agent_toolkits import SQLDatabaseToolkit

toolkit = SQLDatabaseToolkit(db=db, llm=llm)
sql_tools = toolkit.get_tools()
# 内置：sql_db_list_tables / sql_db_schema / sql_db_query_checker / sql_db_query

sql_agent = create_agent(
    llm, sql_tools,
    system_prompt=("先探索 schema，再写只读 SELECT，用 checker 自检、query 限量执行，"
                   "报错就修正，最后给出结论与最终 SQL。"),
)
ans = sql_agent.invoke({"messages": [("user", "华东区销售额 Top3 的品类，并解释结果")]})
print(ans["messages"][-1].content)

官方 SQL 工具的默认提示里已包含关键约束：默认加 LIMIT、不要 SELECT *、执行前 double check、报错就重写——本质就是把第 4–6 章的最佳实践固化成了工具行为。

方式 B：用 LangGraph/LangChain Agent + 自定义工具手写（更可控，Notebook 中完整给出）

# agentic_text2sql_custom.py（骨架）
from langchain_core.tools import tool
from langchain.agents import create_agent   # 等价旧写法 langgraph.prebuilt.create_react_agent

@tool
def list_tables() -> str:
    """列出数据库中所有表及其注释。"""
    return db.get_usable_table_names()

@tool
def describe_table(table_names: str) -> str:
    """查看给定表（逗号分隔）的建表语句、列与外键。"""
    return db.get_table_info(table_names=[t.strip() for t in table_names.split(",")])

@tool
def sample_rows(table: str, n: int = 5) -> str:
    """查看某张表的 n 行样例数据，用于了解列的真实取值格式。"""
    return safe_run(f"SELECT * FROM {table} LIMIT {n}")   # 表名来自白名单，安全

@tool
def run_sql(sql: str) -> str:
    """只读执行 SQL 并返回结果；仅允许 SELECT，自动限量、带报错信息。"""
    ok, msg = static_check(sql)
    return msg if not ok else safe_run(sql)

agent = create_agent(
    llm, [list_tables, describe_table, sample_rows, run_sql],
    system_prompt=("你是数据分析 Agent：先用 list/describe/sample 探索并确认取值，"
            "再写 SQL，用 run_sql 执行，报错就根据错误修正，最多尝试 5 次，"
            "最后给出结论、所用 SQL 和口径说明。"),
)
res = agent.invoke({"messages": [("user", question)]})
print(res["messages"][-1].content)

8.4 Agentic vs Pipeline：本质区别与取舍

维度	固定 Pipeline（4–6 章）	Agentic Text2SQL
流程	开发者写死步骤	模型按问题临场决定探索路径
大库/复杂问题	依赖 schema linking 质量	可多轮看样例、拆子问题，更鲁棒
延迟/成本	可控、较低	轮次多、token 多、延迟高
可预测性	强，便于调试	弱，需要 trace 和护栏
适用	高频、模式固定的标准问数	开放式、探索式、复杂分析

实践建议：先用「Schema Linking + 增强 + 自纠错」的固定 Pipeline 覆盖 80% 标准问题（快、稳、便宜）；只把剩下高复杂度、开放式的问题路由给 Agentic 模式（Adaptive 思想）。

9. 评测体系：EX / EM / VES 与 Spider / BIRD

9.1 两种核心指标

指标	全称	怎么算	优缺点
EM（Exact Match）	精确匹配率	生成 SQL 与标准 SQL 字符串规范化后是否一致	太死板：`SELECT name FROM t` 和 `SELECT t.name FROM t` 等价却不匹配，会低估
EX（Execution Accuracy）	执行正确率	两条 SQL 分别在数据库执行，结果集是否一致	更符合实际、是业界主指标；但可能「错错得正」（不同错误恰好同结果）

为什么主用 EX：SQL 是声明式语言，同一问题有无穷多种等价写法，比对字符串没有意义；「能不能查出正确结果」才是用户真正关心的。工程上还会补充：

VES（Valid Efficiency Score）：BIRD 提出，在 EX 基础上惩罚「能跑对但极其低效」的 SQL（结果对 + 运行效率综合打分）。
组件级指标：schema linking 准确率、SQL 可执行率、首轮正确率、平均纠错轮数（用于定位是「找错表」还是「写错 SQL」）。

9.2 两大权威基准

Spider（2018，耶鲁）：跨领域 Text2SQL 基准，200 个数据库、按 SQL 难度分 Easy/Extra-Hard；训练库和测试库不同，考跨域泛化。偏「干净」的学术设定。
BIRD（NeurIPS 2023）：更贴近真实——大规模、脏数据/噪声取值、需要外部业务知识、关注效率（VES），95 个真实大库。BIRD 上人类数据分析师基线约 92.96% EX，说明真实场景仍很难，模型与人类有差距。

9.3 落地时怎么建自己的评测集

不要只看公开榜。生产中应沉淀自己业务的「问题→标准 SQL→期望结果」黄金集（几十到几百条，覆盖各难度与高频指标），每次改 Prompt/换模型/改 schema linking，都在这个集子上跑 EX，用数据驱动迭代——这和 RAG 要建问答评测集是同一个道理。

10. 生产化：安全、歧义澄清、结果解释与成本

Text2SQL 直接操作数据库，安全是红线，比 RAG 更需要严肃对待。

10.1 安全护栏（必须做）

flowchart LR
    SQL[模型生成 SQL] --> A[① 语句类型白名单<br/>只允许 SELECT]
    A --> B[② 只读账号连接<br/>数据库层面 revoke 写权限]
    B --> C[③ 强制 LIMIT/超时<br/>防止全表扫描拖垮库]
    C --> D[④ 标识符白名单校验<br/>表/列必须真实存在]
    D --> E[⑤ 只读事务 + 副本库执行<br/>隔离生产主库]
    E --> F[⑥ 高风险动作人工审批]

双保险只读：应用层正则拦截非 SELECT 只是第一道；数据库连接必须使用只有 SELECT 权限的账号（纵深防御，模型/Prompt 被绕过也写不了数据）。
防 SQL 注入与标识符注入：值用参数化传递；表名/列名不能直接拼接模型输出，必须过 schema 白名单。
强制限量与超时：自动注入 LIMIT、设置 statement_timeout，在只读副本而非主库执行，避免一条笛卡尔积把生产库打挂。
危险操作显式拦截：ATTACH、PRAGMA、load_extension、多语句（; 后再接语句）、INTO OUTFILE 等一律拒绝。
权限最小化与审计：按用户角色限制可访问的表/行（行级权限），所有生成 SQL 与执行记录留痕，可追溯、可回放。

10.2 歧义澄清：不要让模型「猜」

当问题存在多种合理解读时（「销售额」含不含退款？「本季度」是自然季度还是滚动 90 天？），先反问澄清而不是默默选一个。可以让模型输出结构化的「澄清问题」，由前端向用户确认后再生成——宁可多一轮交互，也不要给出一个看似精确实则错误的数字。

10.3 结果交付：不只是一张表

结果解释：用自然语言说明「这个数字是怎么算出来的」（用了哪些表、过滤条件、口径）。
附带最终 SQL：让懂行的用户可以核验，建立信任。
可视化建议：时间序列→折线、占比→饼图、排名→柱状，让模型同时输出推荐图表类型。
空结果/异常结果的解释：主动说明是「确实没有数据」还是「筛选条件可能有误」。

10.4 成本与延迟优化

Schema linking、表 DDL、指标字典缓存复用；
标准问题走轻量固定 Pipeline，复杂问题才升级 Agent（模型分级：小模型路由/纠错，强模型攻坚）；
给 Agent 设最大迭代轮数与 token 预算；
命中过的「问题→SQL」沉淀为缓存/黄金案例，越用越准、越用越省。

11. 方案对比与选型决策

11.1 五代方案横向对比

方案	核心机制	模型决策	适用库规模/问题	成本	可靠性
Naive Text2SQL	全量 schema 一次生成	无	极小库、简单问题	低	低
+ Schema Linking	先选表列再生成	无（流程固定）	中大库	低	中
+ 上下文增强	注释/口径/few-shot	无	业务术语复杂	低	中高
+ 执行自纠错	执行反馈闭环修正	固定分支	易报错场景	中	高
Agentic Text2SQL	工具化多轮探索	模型自主	大库、复杂/开放分析	高	高（需护栏）

11.2 选型决策树

flowchart TD
    START[要做自然语言查数] --> Q1{库很小(<5张表)且问题简单?}
    Q1 -->|是| N[Naive 即可]
    Q1 -->|否| Q2{表很多/业务术语复杂?}
    Q2 -->|是| SL[Schema Linking + 元数据/口径/Few-shot]
    Q2 -->|否| SL
    SL --> Q3{SQL 经常报错或查空?}
    Q3 -->|是| CR[加执行反馈自纠错闭环]
    Q3 -->|否| Q4{存在复杂多跳/开放式探索分析?}
    CR --> Q4
    Q4 -->|是| AG[升级 Agentic：探索工具集 + ReAct]
    Q4 -->|否| DONE[固定 Pipeline 上线 + 评测集迭代]
    AG --> Q5{有强安全要求?}
    Q5 -->|是| SEC[只读账号+白名单+限量+审批，缺一不可]

11.3 落地顺序建议

先打通 SQLDatabase + 生成 + 只读执行 的最小闭环；
立刻补 Schema Linking + 只读安全护栏（这两件 ROI 最高）；
建数据字典/指标口径库和黄金评测集，用 EX 驱动迭代；
加执行自纠错把可执行率拉满；
最后才对长尾复杂问题上 Agentic，并配 trace、轮数护栏与成本监控。

12. 参考资料

经典论文

Lewis et al.（RAG 原始论文，理解 Text2SQL 与 RAG 的同源思想）, NeurIPS 2020：https://arxiv.org/abs/2005.11401
DIN-SQL: Decomposed In-Context Learning of Text-to-SQL with Self-Correction, NeurIPS 2023：https://arxiv.org/abs/2304.11015
DAIL-SQL, 2023：https://arxiv.org/abs/2308.15363
MAC-SQL: A Multi-Agent Collaboration Framework, 2023：https://arxiv.org/abs/2312.11242
CHESS: Contextual Harnessing for Efficient SQL Synthesis, 2024：https://arxiv.org/abs/2405.16755
BIRD-SQL: A Big Bench for Large-Scale Database Grounded Text-to-SQL Evaluation, NeurIPS 2023：https://arxiv.org/abs/2305.03111 ｜官网 https://bird-bench.github.io/
Spider: A Large-Scale Human-Labeled Dataset for Complex and Cross-Domain Semantic Parsing, EMNLP 2018：https://arxiv.org/abs/1809.08887

官方文档与开源项目

LangChain 官方教程 — Build a SQL agent：https://docs.langchain.com/oss/python/langchain/sql-agent
LangChain SQL Database / SQLDatabaseToolkit API 文档：https://python.langchain.com/docs/tutorials/sql_qa/
Vanna（开源 RAG 式 Text2SQL 框架）：https://github.com/vanna-ai/vanna
WrenAI（开源 AI 数据助手）：https://github.com/Canner/WrenAI
LangGraph 官方文档（状态图、条件边、持久化）：https://docs.langchain.com/oss/python/langgraph/

配套动手材料（同目录）

text2sql_demo.ipynb：可逐格运行——自动建 SQLite 电商库 → Naive 生成 → Schema Linking → Few-shot 增强 → LangGraph 自纠错 → Agentic SQL Agent，已用 deepseek-v4-flash 实测。

requirements.txt、.env.example：依赖与 Key 配置模板。