一、为什么"帮我查下数据"这么难

做数据分析的都知道，业务方最常说的一句话就是"帮我查个数据"。

"上个月华东区退货率最高的三个品类是什么？"——这个问题背后的 SQL 至少涉及 4 张表（订单、退货、商品、区域），2 个 JOIN，1 个子查询和 1 个窗口函数。如果不熟悉 Schema，写出来大概率是错的。

我们平台有 300+ 张业务表，8000+ 个字段。数据分析师每天 60% 的时间在写 SQL、验 SQL、改 SQL。等 SQL 跑对了，已经没精力分析数据了。

所以我们做了一个数据分析 Agent：业务方用自然语言问，Agent 自动生成 SQL→执行→校验→可视化→归因。目标是让分析师从"写 SQL"变成"做决策"。

下面是我们从 58% 准确率一路干到 96% 的完整记录。

阶段	策略	SQL执行成功率	结果准确率	平均SQL复杂度
V1	Few-shot Prompt 直出	71%	58%	低
V2	Schema上下文增强	89%	78%	中
V3	SQL校验+自动修复	97%	89%	中高
V4	归因+多轮记忆	99%	96%	高

二、系统架构

图：NL2SQL Agent 全链路

三、V1：Few-shot 起步（准确率 58%）

# V1: 最基础的 Few-shot NL2SQL
SYSTEM_PROMPT = """你是一个SQL生成助手。根据用户的自然语言查询，生成对应的SQL语句。

数据库是PostgreSQL。可用表：
- orders(id, user_id, product_id, amount, status, created_at, region)
- products(id, name, category_id, price, stock)
- categories(id, name, parent_id)
- returns(id, order_id, reason, created_at)
- users(id, name, region)

示例：
Q: 上个月销售额最高的三个品类
A: SELECT c.name, SUM(o.amount) as total
   FROM orders o JOIN products p ON o.product_id = p.id
   JOIN categories c ON p.category_id = c.id
   WHERE o.created_at >= '2026-07-01' AND o.status != 'cancelled'
   GROUP BY c.name ORDER BY total DESC LIMIT 3

请只返回SQL，不要解释。"""

def query_v1(nl_query):
    response = llm.invoke([
        SystemMessage(content=SYSTEM_PROMPT),
        HumanMessage(content=nl_query)
    ])
    return extract_sql(response.content)

⚠️ V1 的问题：准确率 58%。30% 的错误是因为 Schema 不完整——300 张表只给了 5 张；22% 是 SQL 语法错误（字段名拼错、JOIN 条件缺失）；其余是业务逻辑错误（比如忘了过滤 cancelled 订单）。

四、V2：Schema 动态检索 + 上下文增强（58%→78%）

V1 只给了 5 张表的 Schema，但实际数据库有 300+ 张表、8000+ 字段。LLM 需要先理解"该用哪些表和字段"，然后才能写 SQL。

# V2: Schema动态检索
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 预先将全部表结构和字段描述向量化
# 每条记录格式: "orders表: 订单表，记录所有用户下单信息。字段: amount(金额,decimal)..."

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
schema_store = Chroma(
    collection_name="schema_index",
    embedding_function=embeddings,
    persist_directory="./schema_db"
)

def retrieve_schema(nl_query, top_k=8):
    """根据查询语义，检索最相关的表和字段"""
    # Step 1: 用查询检索相关表
    table_results = schema_store.similarity_search_with_score(
        f"表: {nl_query}", k=min(top_k, 20)
    )

    # Step 2: 对于每张表，检索相关字段
    full_schema = []
    seen_tables = set()
    for doc, score in table_results:
        if score > 0.4:  # 过滤低相关度表
            continue
        table_name = doc.metadata["table_name"]
        if table_name in seen_tables:
            continue
        seen_tables.add(table_name)

        # 检索该表下的相关字段
        field_results = schema_store.similarity_search_with_score(
            f"{nl_query} {table_name}", k=15
        )
        fields = [doc.metadata for doc, score in field_results
                  if doc.metadata["table_name"] == table_name and score < 0.3]

        full_schema.append({
            "table": table_name,
            "description": doc.metadata["description"],
            "columns": [{"name": f["column_name"], "type": f["data_type"],
                         "comment": f["column_comment"]} for f in fields]
        })

    return full_schema

def query_v2(nl_query):
    schema = retrieve_schema(nl_query)

    # 构造 prompt，注入动态检索的 Schema
    schema_text = json.dumps(schema, ensure_ascii=False, indent=2)
    system = f"""可用表及字段（已根据查询语义筛选）：
{schema_text}

重要规则：
1. 金额字段用SUM/AVG时务必过滤 status != 'cancelled'
2. 时间范围查询注意字段类型（date/timestamp）
3. JOIN时确保关联字段类型匹配"""

    response = llm.invoke([
        SystemMessage(content=system),
        HumanMessage(content=nl_query)
    ])
    return response.content

💡 效果：准确率 58%→78%。但有两个新问题：1）schema 检索偶尔漏掉需要的表（召回率约 85%）；2）SQL 语法错误率仍有 11%。

五、V3：SQL 校验 + 自��修复循环（78%→89%）

我们分析 V2 的失败 SQL，发现大量错误是可以自动修复的：拼错的字段名、缺失的 JOIN 条件、WHERE 子句的类型不匹配。

# V3: SQL校验+自动修复（最多3次重试）
import sqlparse
import sqlalchemy as sa

def validate_and_fix_sql(sql, schema_context, max_retries=3):
    """校验SQL语法和语义，失败则交给LLM修复"""
    engine = sa.create_engine("postgresql://...")

    for attempt in range(max_retries):
        # Step 1: 语法检查（用EXPLAIN模拟执行，不实际跑数据）
        try:
            with engine.connect() as conn:
                conn.execute(sa.text(f"EXPLAIN {sql}"))
                conn.commit()
            # 语法通过了
        except Exception as e:
            error_msg = str(e)
            # Step 2: LLM分析错误并修复
            fix_prompt = f"""以下SQL执行失败：
```sql
{sql}
```

错误信息：{error_msg}
数据库Schema：{schema_context}

请分析错误原因并修复SQL。常见问题：
- 字段名拼写错误
- JOIN条件缺失
- 数据类型不匹配（如字符串和数字比较）
- 聚合函数使用错误

只返回修复后的SQL，不要解释。"""
            sql = llm.invoke([HumanMessage(content=fix_prompt)]).content.strip()
            # 去掉可能的 markdown 包装
            sql = sql.replace("```sql", "").replace("```", "").strip()
            continue

        # Step 3: 语义检查 - 验证引用的表和字段是否都存在
        semantic_errors = check_semantic(sql, schema_context)
        if semantic_errors:
            fix_prompt = f"""SQL使用了不存在的表或字段：{semantic_errors}
请根据Schema修正SQL。"""
            sql = llm.invoke([HumanMessage(content=fix_prompt)]).content.strip()
            continue

        # 全部通过
        return sql, True

    return sql, False  # 3次后仍然失败

def check_semantic(sql, schema):
    """检查SQL引用的表和字段是否都在schema中"""
    parsed = sqlparse.parse(sql)[0]
    tokens = [str(t) for t in parsed.tokens if not t.is_whitespace]

    errors = []
    all_columns = set()
    all_tables = set()
    for table_info in schema:
        all_tables.add(table_info["table"])
        for col in table_info["columns"]:
            all_columns.add(f"{table_info['table']}.{col['name']}")
            all_columns.add(col['name'])  # 无前缀引用

    # 简单检查（生产环境应该用sqlglot解析AST）
    for token in tokens:
        if "." in token:
            if token not in all_columns:
                errors.append(f"未知字段: {token}")

    return errors

💡 效果：准确率 78%→89%。修复循环解决了 SQL 语法错误，但语义错误（SQL 能跑但结果不对）仍然是 11% 的主要来源。比如用户问"华东区上月退货率最高的品类"，LLM 写了个 JOIN 了但忘过滤 returns 表，查出的是"销售额最高"而不是"退货率最高"。

六、V4：结果校验 + 多轮对话 + 归因分析（89%→96%）

V3 只校验了 SQL 能不能跑，没校验结果对不对。V4 加了三个关键能力：

# V4: 结果合理性校验 + 多轮对话
def query_v4(nl_query, conversation_history=None):
    # Step 1: 生成SQL（V3管道）
    sql = generate_sql_v3(nl_query)

    # Step 2: 执行SQL并获取结果
    result_df = execute_sql(sql)

    # Step 3: 合理性校验 - 让LLM检查结果是否合理
    check_result = llm.invoke(f"""用户查询：{nl_query}
生成的SQL：{sql}
查询结果（前10行）：{result_df.head(10).to_markdown()}
统计描述：{result_df.describe().to_markdown()}

请检查结果是否合理：
1. 数据量级是否符合预期？
2. 数值范围是否有明显异常？
3. 查询逻辑是否完全匹配用户意图？

如果有问题，说明具体问题并给出修正后的SQL。如果正确，回复'OK'。""")

    if "OK" not in check_result.content:
        # 结果不合理，根据反馈修正
        sql = extract_sql(check_result.content)
        result_df = execute_sql(sql)

    # Step 4: 自动归因分析（不仅给数据，还解释原因）
    attribution = llm.invoke(f"""用户查询：{nl_query}
查询结果：{result_df.to_markdown()}

请基于数据给出归因分析：
1. 主要发现（2-3点）
2. 可能的原因分析
3. 建议的后续行动""")

    # Step 5: 生成可视化建议
    viz_config = llm.invoke(f"""根据以下数据和用户意图，推荐最合适的图表类型：
数据：{result_df.head(3).to_markdown()}
维度：{result_df.columns.tolist()}

返回图表配置JSON：{{"chart_type":"bar/line/pie","x":"列名","y":"列名或列名列表"}}""")

    return {
        "sql": sql,
        "data": result_df.to_dict("records"),
        "attribution": attribution.content,
        "chart": viz_config.content
    }

准确率计算公式：
Accuracy = (正确执行的查询数 / 总查询数) × 100%
其中"正确执行" = SQL语法通过 ∩ 语义正确 ∩ 结果合理性校验通过

七、八个真实坑位

#	坑位	现象	根因	解法
1	多表JOIN错配	查退货率时JOIN了orders和products但忘了returns，查出的是销售额排名	LLM对"退货率"的语义理解停留在表面	V4 结果合理性校验：发现结果中没有returns相关字段，触发修正
2	聚合函数误用	用户问"平均客单价"，LLM生成 AVG(amount) 但没按用户分组	LLM不熟悉业务口径（客单价=SUM/订单数，不是AVG）	在Schema中加入业务口径定义字段
3	时间范围歧义	"上个月"在不同查询中有不同含义（自然月/近30天），导致数据不准	自然语言的模糊性	增加时间消歧模块：强制LLM返回时间范围，与用户确认
4	NULL值处理缺失	WHERE条件中 NULL != 'cancelled' 为NULL（不是TRUE），导致过滤失效	LLM忽略了SQL三值逻辑	在System Prompt中强制注入NULL处理规则
5	子查询性能爆炸	LLM生成的SQL包含5层子查询+NOT IN，执行时间300s+	LLM不理解执行计划，生成低效SQL	加SQL复杂度上限检查（子查询≤2层），超限自动用CTE重写
6	Schema召回不全	某些查询需要的表未被向量检索召回	搜索词和Schema描述之间的语义鸿沟	双路召回：稠密向量 + BM25关键词，Top-10去重后取交集
7	多轮对话上下文丢失	用户第二轮说"再按区域拆分"，Agent不知道上一轮的表	多轮对话缺乏Schema和SQL上下文传递	每轮结果存入对话记忆，新查询自动继承上轮的FROM表
8	结果可视化类型错误	LLM为时间序列数据推荐饼图	LLM缺乏数据可视化最佳实践知识	硬编码规则：时间维→折线图，分类维→柱状图，占比→饼图

八、完整演进数据

指标	V1	V2	V3	V4
SQL执行成功率	71%	89%	97%	99.2%
结果准确率	58%	78%	89%	96%
平均生成耗时	1.1s	2.3s	3.5s	4.2s
平均查询Token数	800	2100	3800	5200
支持表数	5	300+	300+	300+
平均SQL复杂度	2.1 JOINs	3.4 JOINs	4.1 JOINs	4.8 JOINs
修复成功率	-	-	72%	86%

九、环境依赖

组件	版本	用途
Python	3.11.9	运行环境
LangChain	0.3.13	Agent框架
ChromaDB	0.5.5	Schema向量检索
PostgreSQL	16.3	业务数据库
SQLAlchemy	2.0.30	SQL执行+校验
sqlparse	0.5.1	SQL解析
OpenAI API	gpt-4o-mini	LLM推理
ECharts/Apache Superset	-	可视化

十、总结

NL2SQL 的难度排序：语法正确 < 语义正确 < 业务正确。V1-V3 解决了前两个，V4 通过结果合理性校验才摸到第三层的门槛。

1. Schema是最关键的信息源。没有完整的表和字段描述，准确率上限就是 60%。
2. SQL修复循环比一次生成更可靠。与其花时间调 prompt 让它一次写对，不如让它写了再改。86% 的错误可以通过校验→修复两轮解决。
3. 结果合理性校验是最后一道防线。SQL 能跑不代表结果对，必须让 LLM 检查结果是否符合用户意图。

⚠️ 适用边界：本方案适用于有结构化 Schema 的关系型数据库场景。不适用于非结构化数据查询（需要结合 RAG），也不适用于需要实时更新的流式数据（SQL 校验的 EXPLAIN 有秒级延迟）。如果你的表少于 20 张，不需要上向量检索——直接把全部 Schema 放进 Prompt 就行。

参考：LangChain SQL Agent文档(2026-06)、Defog SQLCoder技术报告(2026-04)、淘宝Starrocks NL2SQL实践(2026-05)、CSDN RAG优化12杠杆(2026-07)