企业官网建设流程全解析

带有业务枚举约束的 Text2SQL：基于 JSON Schema 校验引导模型输出合法 SQL

在企业级数据中台与智能报表系统中，将自然语言转换为可执行 SQL（Text2SQL）是极其核心的交互能力。然而，直接让大型语言模型（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'

当数仓表结构扩展到几十张表、上百个枚举状态机时，这种做法面临崩溃：

1. 注意力漂移与遗忘：随着 Context 长度拉长，大模型对于后半段复杂复合过滤条件的注意力权重衰减，极易使用训练语料中的通用词替代专用业务编码。
2. 多值过滤笛卡尔爆炸：当业务提问包含“非微信但交易成功”等否定和集合操作时，模型自发构造的NOT IN (...)常常漏掉隐藏的软删除状态枚举。
3. 安全注入隐患：未经校验的字符串直接嵌入 SQL 片段，破坏了预编译参数化执行的基本原则。

两阶段生成架构：中间语义规约（Query Spec）

消除非确定性的最佳手段是消除自由文本生成。我们拒绝让大语言模型直接吐出最终的SELECT ... FROM字符串，而是引入面向编译器设计的中间语义层——结构化查询规约（Query Spec）。

执行链路分为两步：

1. 语义解构：引导模型输出严格遵守特定业务 JSON Schema 的 JSON 对象。在该 Schema 中，所有的过滤字段及其比较值必须满足预定义的enum限制。
2. 确定性代数渲染：由本地确定性代码解析校验该 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_factory=list, description="过滤条件列表") limit: Optional[int] = Field(100, ge=1, le=1000, 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 = f"SELECT {select_clause} FROM {spec.target_table}" params = [] where_clauses = [] for condition in spec.filters: if condition.field not in valid_fields: raise ValueError(f"WHERE 条件包含非法字段: {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被拦截后，系统不应简单直接报错打回。常见的恢复策略有两种：

1. Schema 错误上下文重试（Reflection Loop）：将 Pydantic 抛出的明确错误堆栈回传给模型：“输入值 'PAID' 不在合法枚举集合 ['PAY_SUCCESS', 'CLOSED', ...] 中，请纠正并重新生成”。依靠大模型的自修正能力，二次调用的纠偏成功率超过 95%。
2. 轻量向量映射（Vector Normalization）：在本地渲染层维护一个小型词嵌入索引。当模型传入未定义的枚举别名时，在特定字段的枚举库中执行快速余弦相似度匹配。若相似度超过 0.88（例如PAID极度逼近PAY_SUCCESS），在记录告警日志的前提下完成隐式转换，兼顾严格性与用户体验。

通过在架构前端截断 SQL 自由拼接，利用 JSON Schema 进行强类型规约，整个 Text2SQL 系统的数据流从“不确定性的概率猜测”收敛为“确定性的编译器管道”，彻底清除了破坏数据一致性与数据库安全的核心隐患。