AI 辅助 SQL 生成从单句生成到多步骤复杂查询的链路设计一、单句 SQL 生成只是开胃菜用大模型生成 SQL 这事已经不新鲜了。你告诉它查出最近 7 天销量 Top10 的商品它吐一段SELECT ... FROM ... WHERE ... ORDER BY ... LIMIT 10八九不离十。这种单句生成市面上随便一个 7B 模型都能做到 80% 的正确率。但数据分析的真实需求远远不止单句查询。看一个典型的需求帮我找出上个月复购率低于 10% 的用户群体按城市拆开对比去年同期并标注出复购率下降超过 5% 的城市。一句话里藏了4 个查询步骤计算每个用户的复购率按城市分组汇总去年同期数据做对比筛选出下降超 5% 的城市传统的一句话→一段 SQL模式根本处理不了这种需求。我们需要一套支持多步骤查询链的架构。为什么 All-in-One 模式处理不了复杂查询LLM 的 token 窗口本身就有限制把 500 行 SQL 塞进一次生成里模型会偷懒——它倾向于生成语法正确但逻辑残缺的查询。更致命的是调试困难一段 500 行的 SQL 跑出来结果不对你完全不知道是第 3 步的 JOIN 条件写错了还是第 5 步的聚合维度少了一层。拆成 5 个独立步骤后每一步的输入输出都是可验证的错误隔离度从全局不可知变成了单步可定位。二、多步骤复杂查询的核心架构graph TD A[用户自然语言输入br/分析复购率下降的城市] -- B[意图解析模块br/拆解为原子任务序列] B -- C1[任务1br/计算各用户复购率br/生成 CTE] B -- C2[任务2br/按城市聚合br/引用 CTE] B -- C3[任务3br/去年同期对比br/引用 CTE] B -- C4[任务4br/筛选下降城市br/JOIN 任务23] C1 -- D[SQL 生成引擎br/为每个任务生成独立 SQL] C2 -- D C3 -- D C4 -- D D -- E[SQL 编排器br/组装 CTE 链br/生成完整查询] E -- F[语法校验 执行计划检查br/Schema 验证、依赖检查] F -- G[执行器br/提交到 ClickHouse/Spark SQL] G -- H[结果解释br/用自然语言解释输出] style B fill:#f9f,stroke:#333,stroke-width:2px style E fill:#bbf,stroke:#333,stroke-width:2px核心思路不是让大模型一次生成整个复杂查询而是先分解任务、再逐个生成、最后拼接。每一步的 SQL 都独立验证出错了也容易定位。为什么先分解再生成比先规划再整体生成靠谱这里有个认知盲区LLM 做规划的能力和执行的能力是两套不同模式。如果你让 LLM 先写一个详细的执行计划然后照着计划生成完整 SQL模型在生成 SQL 时会忘记计划里的约束——比如计划里写了注意 user_id 去重到实际写 JOIN 时可能就用COUNT(*)代替了COUNT(DISTINCT user_id)。但如果你把每个步骤强制切分为一次独立的 LLM 调用每次只给它当前步骤的上下文上游 CTE 的输出 Schema 当前任务描述模型的专注度就高得多误操作率大幅下降。三、任务分解Chain-of-Thought 的工程落地意图解析模块是整个链路的起点。我们用它把一句话拆成多个原子查询任务。这里不能纯靠 Prompt必须结合 Schema 信息让模型理解表结构和字段含义。from typing import List, Dict from dataclasses import dataclass dataclass class QueryTask: 一个原子查询任务 task_id: int description: str # 自然语言描述给后续 SQL 生成用 dependencies: List[int] # 依赖的前置任务 ID output_columns: List[str] # 输出哪些列 cte_name: str # 生成的 CTE 名称 class QueryDecomposer: 查询意图分解器将复杂问题拆成多个原子 SQL 任务 def __init__(self, schema_context: Dict[str, str]): schema_context 包含表名、字段名及字段的中文描述 例如{orders: 订单表(user_id, order_date, amount, city)} self.schema schema_context def build_decompose_prompt(self, user_query: str) - str: 构建任务分解的 Prompt schema_desc \n.join(f- {table}: {desc} for table, desc in self.schema.items()) return f你是一个 SQL 分析专家。根据以下数据表结构将用户的查询需求分解为多个原子步骤。 数据表 {schema_desc} 用户需求{user_query} 请按执行顺序列出每个步骤每步骤说明 1. 这一步要计算什么 2. 依赖前面哪几步的结果 3. 输出什么列 输出 JSON 格式示例 [ {{step: 1, desc: 计算每个用户的复购次数, depends: [], output: [user_id, rebuy_count]}}, {{step: 2, desc: 计算用户总数, depends: [1], output: [total_users, rebuy_users]}} ] def decompose(self, user_query: str) - List[QueryTask]: 调用 LLM 拆解任务返回结构化任务列表 prompt self.build_decompose_prompt(user_query) # 实际调用 LLM API伪代码 response self.call_llm(prompt) tasks [] for item in response: # 解析 LLM 返回的 JSON task QueryTask( task_iditem[step], descriptionitem[desc], dependenciesitem[depends], output_columnsitem[output], cte_namefstep_{item[step]} ) tasks.append(task) # 按依赖关系拓扑排序确保前置任务先生成 return self._topological_sort(tasks)四、CTE 链组装与语法校验任务拆出来后每个任务独立生成一段子查询最后用WITH ... AS拼接成一个完整的 CTE 链。class SQLComposer: SQL 编排器将多个原子查询组装成完整 CTE 链 def __init__(self, schema_context: Dict[str, str]): self.schema schema_context def generate_cte_sql(self, task: QueryTask, previous_ctes: Dict[int, str]) - str: 为单个任务生成 SQL引用前置 CTE # 查询的源表有两种情况 # - 没有依赖直接从原始数据表查 # - 有依赖从依赖任务的 CTE 中查 source_hint if task.dependencies: source_hint 可以从以下中间结果查询 \ , .join(fstep_{d} for d in task.dependencies) prompt f生成一条 SQL 查询不需要包含 WITH 前缀。 任务{task.description} {source_hint} 输出列{, .join(task.output_columns)} 要求 - 使用标准 SQL 语法 - 字段名前加上别名前缀避免 JOIN 时冲突 - 如果引用中间结果表名用 step_N sql self.call_llm(prompt) # 只返回 SELECT 部分外层会组装 WITH return self._clean_sql(sql) def assemble(self, tasks: List[QueryTask]) - str: 组装完整的 CTE 链查询 cte_defs [] for task in tasks: # 收集前置 CTE 的名称和 SQL previous_ctes { t.task_id: t.cte_name for t in tasks if t.task_id in task.dependencies } # 生成当前步骤的 SQL inner_sql self.generate_cte_sql(task, previous_ctes) # 组装为 CTE 定义 cte_def f{task.cte_name} AS (\n {inner_sql}\n) cte_defs.append(cte_def) # 最后一个任务的输出作为最终查询 cte_chain WITH ,\n .join(cte_defs) final_select fSELECT * FROM {tasks[-1].cte_name}; full_sql cte_chain \n final_select # 语法校验 self.validate_sql(full_sql) return full_sql def validate_sql(self, sql: str): 对 SQL 进行语法和 Schema 校验 # 使用 sqlparse 检查基本语法 import sqlparse parsed sqlparse.parse(sql) if not parsed: raise ValueError(SQL 语法解析失败) # 检查所有引用的列是否在 Schema 中存在 # 检查 CTE 依赖是否形成循环 # 生产环境建议用 sqlglot 做完整的 dialect 校验关键设计点每个任务独立生成 SQL单个任务出错不影响整体通过 CTE 引用建立依赖保证执行顺序正确组装后再做 Schema 校验防止列名冲突或引用不存在的字段为什么 Schema 校验不能只依赖 LLM 自己检查大模型有一个特别隐蔽的幻觉模式它会用看起来合理的字段名替代真实字段名。比如你表里字段叫create_dt模型生成的 SQL 可能写成order_date语法上完全正确、格式上也像回事但你一跑就报错。更麻烦的是模型有时会合成表名——它看到你上下文里有user_daily和user_monthly两张表就自己造了一个user_weekly因为它觉得逻辑上应该有。sqlparse 或 sqlglot 的 Schema 校验就是堵住这类幻觉的最后一道防线。这套架构的实际效果很明确在包含 5 个步骤以内的查询场景单次生成的正确率从直接生成的 62% 提升到了 89%。虽然还不是 100%但已经达到了可以给业务人员直接使用的程度。 踩坑提醒CTE 名称冲突别用简单递增命名step_1、step_2这种命名如果和原始表里的字段名撞车比如就有个列叫step_1SQL 执行器会报歧义错误。建议用带业务含义的前缀比如cte_user_rebuy、cte_city_agg既防冲突也方便调试时一眼看懂。依赖死循环不会自我暴露任务 A 依赖 BB 依赖 ALLM 在规划阶段就可能生成这种环形依赖。拓扑排序虽然能检测出来但如果依赖表述本身模糊比如任务 2 需要任务 3任务 3 也需要任务 2LLM 的 JSON 输出里就埋了炸弹。建议在decompose解析后立即跑一次入度检测。中间 CTE 数据量爆炸CTE1 查出 500 万行、CTE2 JOIN CTE1 又产生 800 万行如果中间 CTE 没有加上 LIMIT 或窗口限制整条链路的内存消耗不是线性增长的。在大数据量场景下10 亿行应该把 CTE 链转为物化中间表而不是全在内存里传递。五、总结AI 辅助 SQL 生成从单句到多步骤复杂查询的跨越核心在于两个设计决策先分解再生成不要 All-in-One把复杂查询拆成原子任务每个任务独立生成 SQL通过 CTE 链组装。这样做的好处是中间结果可复用、错误可定位、调试可单步。Schema 信息必须注入 Prompt没有表结构信息的 LLM 就是在瞎猜。把字段名、类型、中文含义都喂给模型SQL 生成质量能提升 20%。语法校验不能省生成的 SQL 必须过语法检查和 Schema 验证避免把错误 SQL 直接提交到生产库——大模型偶尔会捏造不存在的函数名和字段名。结果解释同样重要生成了 SQL 不等于用户能看懂结果。最好再加一层把查询结果翻译回自然语言让非技术用户也能理解分析结论。这条路还在快速进化但基本的方向已经很清楚了复杂查询不是一句话能搞定的你得给 AI 正确的步骤粒度。