带有业务枚举约束的 Text2SQL基于 JSON Schema 校验引导模型输出合法 SQL在企业级数据中台与智能报表系统中将自然语言转换为可执行 SQLText2SQL是极其核心的交互能力。然而直接让大型语言模型LLM生成原生 SQL 语句时最常遭遇且隐蔽性极高的问题是业务枚举值的“合规性幻觉”。例如底层交易表的order_status字段在数仓中严格定义为[PAY_SUCCESS, CANCELLED, REFUNDED]但模型在理解“查询已付款的订单”时极易生成WHERE order_status PAID或WHERE order_status SUCCESS。这类 SQL 在语法层面无懈可击数据库解析器能顺利执行但最终返回空数据集造成业务指标静默归零。若涉及更复杂的机构编码、渠道标识或租户状态随意拼接的枚举值更会导致跨分区全表扫描甚至拖垮整个 OLAP 引擎。本文给出一种基于严格 JSON Schema 校验的两阶段 Text2SQL 架构通过结构化中间表达与确定性校验器杜绝非法枚举值渗入生产数据库。朴素 Prompt 约束的失控根源工程团队在面对枚举值幻觉时通常的直觉反应是在 Prompt 中硬编码元数据-- 糟糕的提示词工程示例 请根据用户问题生成 SQL。 表 schema: orders (id, user_id, status, pay_channel, amount) 注意 status 字段只能是: WAIT_PAY, PAY_SUCCESS, CLOSED, REFUND_APPLY, REFUND_DONE pay_channel 字段只能是: WX_PAY, ALIPAY, UNIONPAY, APPLE_PAY当数仓表结构扩展到几十张表、上百个枚举状态机时这种做法面临崩溃注意力漂移与遗忘随着 Context 长度拉长大模型对于后半段复杂复合过滤条件的注意力权重衰减极易使用训练语料中的通用词替代专用业务编码。多值过滤笛卡尔爆炸当业务提问包含“非微信但交易成功”等否定和集合操作时模型自发构造的NOT IN (...)常常漏掉隐藏的软删除状态枚举。安全注入隐患未经校验的字符串直接嵌入 SQL 片段破坏了预编译参数化执行的基本原则。两阶段生成架构中间语义规约Query Spec消除非确定性的最佳手段是消除自由文本生成。我们拒绝让大语言模型直接吐出最终的SELECT ... FROM字符串而是引入面向编译器设计的中间语义层——结构化查询规约Query Spec。执行链路分为两步语义解构引导模型输出严格遵守特定业务 JSON Schema 的 JSON 对象。在该 Schema 中所有的过滤字段及其比较值必须满足预定义的enum限制。确定性代数渲染由本地确定性代码解析校验该 JSON 对象完成严格的类型断言最后通过抽象语法树AST安全渲染出最终的目标 SQL。[自然语言输入] │ ▼ [大模型 Constrained Generation] │ (受控解码强制输出 JSON) ▼ [JSON Schema 拦截校验器] ──(校验失败)── [立即熔断/上下文重试] │ ▼ (校验通过) [确定性 AST 参数化渲染器] │ ▼ [合法 SQL 与预编译参数] ── [OLAP/OLTP 数据库引擎]核心实现Schema 约束与参数化解析器以下代码给出使用 Python 3 标准库及pydantic实现的完整工业级防线演示如何将自然语言的语义映射到强类型枚举体系中。import json from enum import Enum from typing import List, Optional, Union from pydantic import BaseModel, Field, ValidationError # 1. 严格定义数仓底层的业务状态枚举 class OrderStatusEnum(str, Enum): WAIT_PAY WAIT_PAY PAY_SUCCESS PAY_SUCCESS CLOSED CLOSED REFUND_APPLY REFUND_APPLY REFUND_DONE REFUND_DONE class PayChannelEnum(str, Enum): WX_PAY WX_PAY ALIPAY ALIPAY UNIONPAY UNIONPAY APPLE_PAY APPLE_PAY class OperatorEnum(str, Enum): EQ NEQ ! IN IN NOT_IN NOT_IN # 2. 构造支持严密递归的过滤表达式树 class FilterCondition(BaseModel): field: str Field(..., description过滤字段名) operator: OperatorEnum Field(..., description比较运算符) value: Union[OrderStatusEnum, PayChannelEnum, int, float, List[Union[OrderStatusEnum, PayChannelEnum, str]]] Field( ..., description比较值若为枚举字段必须严格属于业务枚举集 ) class QuerySpec(BaseModel): target_table: str Field(..., description目标业务表名) selected_fields: List[str] Field(..., description查询产出字段) filters: List[FilterCondition] Field(default_factorylist, description过滤条件列表) limit: Optional[int] Field(100, ge1, le1000, description安全拉取行数) # 3. 确定性 SQL 渲染器 class SecureSQLRenderer: ALLOWED_TABLES {orders, order_items} FIELD_WHITE_LIST { orders: {id, user_id, order_status, pay_channel, amount, create_time} } classmethod def render(cls, spec: QuerySpec) - (str, list): # 表名防注入白名单拦截 if spec.target_table not in cls.ALLOWED_TABLES: raise ValueError(f非法目标表访问: {spec.target_table}) valid_fields cls.FIELD_WHITE_LIST[spec.target_table] # 字段名白名单拦截 for col in spec.selected_fields: if col not in valid_fields: raise ValueError(f未知列字段: {col}) select_clause , .join(spec.selected_fields) sql fSELECT {select_clause} FROM {spec.target_table} params [] where_clauses [] for condition in spec.filters: if condition.field not in valid_fields: raise ValueError(fWHERE 条件包含非法字段: {condition.field}) # 处理枚举字段与值的一致性校验 val condition.value if isinstance(val, Enum): val val.value elif isinstance(val, list): val [item.value if isinstance(item, Enum) else item for item in val] if condition.operator in (OperatorEnum.IN, OperatorEnum.NOT_IN): if not isinstance(val, list) or len(val) 0: raise ValueError(f操作符 {condition.operator.value} 要求值必须为非空数组) placeholders , .join([?] * len(val)) where_clauses.append(f{condition.field} {condition.operator.value} ({placeholders})) params.extend(val) else: where_clauses.append(f{condition.field} {condition.operator.value} ?) params.append(val) if where_clauses: sql WHERE AND .join(where_clauses) sql f LIMIT {spec.limit} return sql, params # 运行测试 if __name__ __main__: # 场景 A: 模型受到 Schema 约束输出了完全合法的 JSON 报文 llm_legal_payload { target_table: orders, selected_fields: [id, amount], filters: [ { field: order_status, operator: , value: PAY_SUCCESS }, { field: pay_channel, operator: IN, value: [WX_PAY, ALIPAY] } ], limit: 50 } spec QuerySpec.model_validate_json(llm_legal_payload) safe_sql, sql_params SecureSQLRenderer.render(spec) print(解析成功 SQL:, safe_sql) print(绑定参数:, sql_params) # 场景 B: 模型产生幻觉使用了外部别名 PAID llm_hallucinated_payload { target_table: orders, selected_fields: [id], filters: [ { field: order_status, operator: , value: PAID } ] } try: QuerySpec.model_validate_json(llm_hallucinated_payload) except ValidationError as err: print(\n拦截到非法枚举幻觉触发熔断:) print(err.errors()[0][msg])生产落盘中的工程权衡与实战策略1. 动态元数据与 JSON Schema 的实时注入在大型企业中字典表通常存储在 MySQL 元数据库或配置中心如 Nacos/Apollo中且存在日常运维的增删改。硬编码 Pydantic 类只适用于小型系统。工业级落地解法建立后台定时同步任务拉取数据字典表的变更事件通过 Binlog 或 CDC动态编译生成 JSON Schema 定义并将其序列化存入 Redis。当 Text2SQL 网关接收到请求时根据识别到的实体表名动态加载对应的 JSON Schema 结构传递给下游的大模型驱动层如利用 OpenAI Structured Outputs 的response_format机制或者 vLLM / llama.cpp 的 Grammars 引擎强制按语法采样。这样不仅保证了 100% 的字段与枚举命中率还避免了全量 Prompt 上下文过载。2. 容错回退机制严格拦截 vs 语义对齐当用户输入“查一下微信支付成功的单子”模型输出非法枚举SUCCESS被拦截后系统不应简单直接报错打回。常见的恢复策略有两种Schema 错误上下文重试Reflection Loop将 Pydantic 抛出的明确错误堆栈回传给模型“输入值 PAID 不在合法枚举集合 [PAY_SUCCESS, CLOSED, ...] 中请纠正并重新生成”。依靠大模型的自修正能力二次调用的纠偏成功率超过 95%。轻量向量映射Vector Normalization在本地渲染层维护一个小型词嵌入索引。当模型传入未定义的枚举别名时在特定字段的枚举库中执行快速余弦相似度匹配。若相似度超过 0.88例如PAID极度逼近PAY_SUCCESS在记录告警日志的前提下完成隐式转换兼顾严格性与用户体验。通过在架构前端截断 SQL 自由拼接利用 JSON Schema 进行强类型规约整个 Text2SQL 系统的数据流从“不确定性的概率猜测”收敛为“确定性的编译器管道”彻底清除了破坏数据一致性与数据库安全的核心隐患。