大模型驱动的 Text2SQL从自然语言到可执行 SQL 的落地架构一、自然语言查询的鸿沟为什么业务方离不开 Text2SQL业务分析师想从数仓取数却写不出一条正确的 SQL。这是数据平台最常见的痛。传统方案是排期提需求等数据团队开发链路长、响应慢。Text2SQL 的目标是把上周华东区客单价最高的三个品类这类自然语言直接转成可执行的 SQL。但生产环境的 Text2SQL 远比 Demo 复杂。表有上百张字段命名晦涩语义歧义严重。用户可能指注册用户也可能指付费用户。单纯的 prompt 拼接无法解决这些问题。这篇文章拆解一套可落地的 Text2SQL 架构覆盖语义层、检索增强与执行校验三个关键环节。更隐蔽的问题在评测环节。离线刷榜的准确率和生产环境的可用率往往是两回事。当表名相似、字段含义重叠时模型生成的 SQL 看似合理却查错了对象。没有一套基于真实业务问法构建的评测集Text2SQL 永远停留在玩具阶段经不起线上复杂语义的检验。二、分层架构语义层、检索增强与校验闭环做 Text2SQL 不能只想着调一个大模型真正要搭的是一套围绕元数据组织的流水线。语义层先把数据库 schema、业务术语、字段注释沉淀成可检索的知识。检索增强在每次查询时把最相关的表和字段注入 prompt约束生成空间。执行校验则用真实数据库做语法与结果验证对失败查询做自动重试或改写。下面是完整的数据流flowchart TB Q[自然语言问题] -- P[问题理解br/意图与实体识别] P -- R[语义检索br/Schema Retriever] R -- K[(元数据知识库br/表/字段/注释)] K -- R R -- G[SQL 生成器br/LLM 约束 Prompt] G -- E[执行引擎br/沙箱数据库] E --|成功| V[结果返回] E --|失败| C[错误诊断br/异常归类] C -- G C -- H[人工兜底br/低置信度告警] V -- V2[结果格式化] style G fill:#e1f5fe style K fill:#fff3e0 style E fill:#f3e5f5语义检索的质量直接决定上限。如果检索到的表错了模型再强也生成不出正确 SQL。因此知识库必须包含表的中文注释、字段的业务含义以及常见的 join 路径。三、生产级实现带检索增强与重试的 SQL 生成下面给出一个可运行的实现骨架包含 schema 检索、带超时的执行与失败重试import asyncio from dataclasses import dataclass from typing import Optional dataclass class QueryContext: 单次查询的上下文 question: str top_tables: list[str] generated_sql: Optional[str] None confidence: float 0.0 class SchemaRetriever: 语义检索根据问题召回最相关的表与字段 def __init__(self, metadata_store, embed_fn, top_k: int 5): self._store metadata_store self._embed embed_fn self._top_k top_k async def retrieve(self, question: str) - list[str]: try: vec await self._embed(question) hits await self._store.search(vec, limitself._top_k) return [h[table] for h in hits] except Exception as e: # 检索失败时退化为全表枚举避免整体不可用 print(fschema 检索失败启用降级: {e}) return await self._store.list_all_tables() class SQLGenerator: SQL 生成器约束 prompt 模型调用 def __init__(self, llm_chat, retriever: SchemaRetriever): self._llm llm_chat self._retriever retriever async def generate(self, question: str) - QueryContext: tables await self._retriever.retrieve(question) prompt self._build_prompt(question, tables) sql await self._llm.chat(prompt, timeout15) return QueryContext(questionquestion, top_tablestables, generated_sqlsql) def _build_prompt(self, question: str, tables: list[str]) - str: schema \n.join(tables) return f基于以下表结构回答\n{schema}\n问题{question}\n只输出 SQL。 class SQLExecutor: 执行引擎沙箱执行 失败重试与诊断 def __init__(self, db_pool, max_retry: int 2): self._pool db_pool self._max_retry max_retry self._llm None # 由外部注入用于错误改写 async def execute(self, ctx: QueryContext) - dict: for attempt in range(self._max_retry 1): try: async with self._pool.acquire() as conn: rows await conn.fetch(ctx.generated_sql) return {ok: True, rows: rows, sql: ctx.generated_sql} except Exception as e: if attempt self._max_retry: return {ok: False, error: str(e), sql: ctx.generated_sql} # 诊断错误后尝试让模型改写 SQL ctx.generated_sql await self._rewrite(ctx, str(e)) return {ok: False, error: unknown} async def _rewrite(self, ctx: QueryContext, error: str) - str: prompt fSQL 执行报错{error}\n原 SQL{ctx.generated_sql}\n请修正。 return await self._llm.chat(prompt, timeout15)这段代码的关键不在模型调用而在三处容错检索失败降级、执行超时控制、错误自动改写。缺少任何一处Text2SQL 在 production 都会频繁翻车。四、边界与权衡Text2SQL 不是银弹Text2SQL 的准确率高度依赖知识库质量。schema 注释缺失时模型只能靠猜复杂多表 join 的准确率会断崖式下跌。执行安全是另一条红线。自然语言可能生成 DELETE 或 UPDATE必须在执行层做只读校验只允许 SELECT并对结果行数设上限防止全表扫描拖垮数仓。成本也不可忽视。每次查询都走大模型高频场景下 token 开销巨大。可行的优化是给高频问法做 SQL 缓存命中后直接复用绕过模型。最后Text2SQL 解决不了口径不一致问题。同一个活跃用户不同部门定义不同。这需要在语义层固化指标口径而不是交给模型自由发挥。多轮对话也是难点。用户常先问按天统计销量再补一句只看华东区。这类指代省略需要系统维护对话状态把上轮的约束带入下轮。如果每次都独立生成上下文就丢了。可行的做法是维护一个累积的查询上下文对象在每轮生成时显式注入历史约束。五、总结Text2SQL 的落地关键不在模型本身而在围绕元数据的工程化封装。语义检索决定上限执行校验兜住下限重试与降级保障可用性。只读约束与结果限流则守住安全边界。建议从高频、口径明确的取数场景切入先把知识库注释补全再逐步扩展到复杂查询。缓存高频问法可显著降低线上成本。