1. 项目概述为什么 Text2SQL 的“最后一公里”必须是 SQL 检查Text2SQL 这个词最近两年在数据产品、BI 工具和低代码平台里出现频率高得离谱但凡带点“自然语言查数据”功能的产品背后几乎都挂着这个标签。可我干了八年数据平台开发从最早手写规则引擎到后来搭基于模板的生成器再到如今用大模型微调 pipeline最常被业务方堵在茶水间问的一句话从来不是“模型准不准”而是“你这生成的 SQL真敢直接跑生产库吗”——这句话就是“SQL 检查”存在的全部理由。它不是锦上添花的后处理模块而是 Text2SQL 流程中真正卡住脖子的“安全阀”。没有它再高的 BLEU 分数、再漂亮的执行准确率EXE Acc都是纸老虎。我亲眼见过一个金融风控看板模型把“近30天逾期率最高的5个客户”翻译成SELECT TOP 5 * FROM customers ORDER BY overdue_rate DESC表面看语法全对但表里根本没有overdue_rate字段而是overdue_days和is_overdue两个布尔数值组合字段更致命的是TOP 5在 PostgreSQL 里根本报错而该系统底层数据库恰恰是 PG。结果就是前端白屏、日志刷屏、DBA半夜被叫醒——问题不在模型没学好而在生成之后没人给 SQL “做个体检”。所谓 SQL 检查核心就干三件事语法合法性校验、语义合理性判断、执行安全性兜底。它不负责把“查销售额”变成SELECT SUM(amount) FROM orders那是 Text2SQL 模型的事它只负责盯着这句生成出来的 SQL像老会计审凭证一样逐字逐句看它有没有拼写错误、字段是否存在、聚合是否漏了 GROUP BY、WHERE 条件会不会导致全表扫描、有没有危险的DROP TABLE或;分隔符注入痕迹。热搜词里反复出现的 AST抽象语法树、正则表达式正是实现这三重检查最常用也最可靠的两把刀AST 负责深度语义解析正则负责快速模式拦截。这不是炫技而是工程落地的刚需——模型会幻觉人会疏忽但检查逻辑只要写对一次就能千次万次守住底线。适合谁读如果你正在搭建内部 BI 助手、客服知识库的问答后端、或者给销售 CRM 加自然语言搜索那你绕不开这个环节如果你是算法工程师只管模型指标不管上线风险那这篇内容可能让你少背两次 P0 故障的锅如果你是 DBA 或数据治理同学看到“SQL 注入”“慢 SQL 优化”这些热词频繁出现在搜索列表里说明业务侧已经把生成式 SQL 当成日常操作了而你的职责就是让这种“日常”不变成“事故现场”。2. 核心思路拆解为什么必须分层检查而不是只靠 AST 或只靠正则很多人第一次做 SQL 检查直觉就是“拿正则扫一遍拦掉DROP|DELETE|;就完事”。我试过上线三天就被打脸。有一次正则写了r(DROP|DELETE)\sTABLE结果业务提了个需求“帮我删掉测试环境里所有以 tmp_ 开头的临时表”模型生成了DROP TABLE tmp_user_log_202405; DROP TABLE tmp_order_cache_202405;——正则只匹配到第一个DROP TABLE第二个被分号隔开直接逃逸。更麻烦的是有些合法 SQL 天然带分号比如 SQL Server 的批处理语句、或某些 ORM 框架生成的多语句脚本。单纯靠字符串匹配就像用筛子捞沙漏点太多还容易误伤。反过来有人迷信 AST觉得“解析成树就万事大吉”。我也踩过坑。用 Python 的sqlglot库把 SQL 解析成 AST 后遍历节点检查字段存在性结果发现SELECT a.name, b.age FROM users a JOIN orders b ON a.id b.user_id这种语句AST 里a.name是个Column节点但它的table属性是aname属性是name要确认name字段在users表里是否存在得先根据别名a反查到原始表users再查users的 schema。这需要维护完整的表-列映射关系而实际业务中schema 是动态的新表随时加字段随时改甚至跨库 JOIN比如sales.customers和logistics.orders。光有 AST 结构没有实时 schema 上下文检查就是空中楼阁。所以真正稳健的 SQL 检查必须是三层嵌套结构第一层是正则快筛层像安检门的 X 光机50ms 内扫出明显违规模式比如;\s*(CREATE|ALTER|DROP|DELETE|UPDATE)、EXEC\s\w、xp_cmdshellSQL Server 特有危险函数、UNION\sSELECT\s.*\sFROM\ssys\.系统表探测等。这一层不求 100% 精准但求 99% 拦截且绝对不能影响主流程性能。第二层是AST 语义层像专业医生做 CT对通过快筛的 SQL 做深度解析。重点检查所有Column节点的table和name是否在已知 schema 中存在需传入当前查询上下文的 database schema 列表GROUP BY子句是否覆盖了所有非聚合字段避免 MySQL 5.7 严格模式报错ORDER BY引用的字段是否在SELECT列表中某些数据库如 Presto 要求严格匹配LIMIT/TOP是否与目标数据库方言兼容如LIMIT 10 OFFSET 20在 SQL Server 2012 才支持旧版得用ROW_NUMBER()。第三层是执行前模拟层可选但强烈推荐像手术前的最终签字。不真执行但用数据库的EXPLAIN或DESCRIBE命令获取执行计划检查是否有Seq Scan全表扫描、Nested Loop笛卡尔积风险、Hash Join内存溢出预警等高危操作符。这一层依赖数据库连接通常放在异步任务或预执行队列里不阻塞用户请求但为高风险 SQL 提供二次确认。这三层不是并列选项而是流水线正则失败直接拒掉AST 失败返回具体错误如“字段 ‘user_name’ 在表 ‘customers’ 中不存在”模拟层异常则标记为“需人工复核”。我在线上系统里实测过三层叠加后非法 SQL 拦截率从单层正则的 82% 提升到 99.7%且平均检查耗时控制在 120ms 内含网络延迟完全满足交互式查询的体验要求。3. 核心细节解析AST 解析与正则设计的实战要点3.1 AST 解析选型、构建与关键节点遍历逻辑AST 解析不是黑盒选对工具和理解其局限性比盲目堆代码重要得多。目前主流开源方案有三个sqlparse纯 Python轻量但不支持方言、sqlglot新兴明星支持 15 方言AST 结构清晰、pyparsing灵活但需手写语法规则学习成本高。我团队线上用的是sqlglot原因很实在它能把SELECT * FROM t1 JOIN t2 USING(id)和SELECT * FROM t1 INNER JOIN t2 ON t1.id t2.id解析成完全一致的 AST 结构极大简化了 JOIN 逻辑的统一检查而且它内置了Dialect模块能自动处理LIMIT/TOP/ROWNUM的方言转换不用我们自己写 if-else。构建 AST 的第一步永远是指定目标方言。sqlglot.parse(SELECT TOP 5 name FROM users, dialecttsql)和sqlglot.parse(SELECT name FROM users LIMIT 5, dialectpostgres)解析出的 AST 根节点类型不同前者是Select下挂Limit后者是Select下挂Limit但Limit的expression类型是Literal而非Number不指定方言后续遍历会出错。我们线上配置了一个dialect_map字典按用户选择的数据库类型如sqlserver2019,mysql8,postgresql14映射到sqlglot的标准方言名确保解析一致性。关键节点遍历核心是抓住四个“命脉节点”Column节点检查字段存在性。遍历时先取node.table可能是别名再通过别名反查原始表名node.find(sqlglot.exp.Table)最后查 schema。注意node.name是字段名但可能带引号如user-name需去掉引号再查node.table可能为空无表前缀此时需默认查当前默认 schema 的所有表。Group节点检查GROUP BY完整性。遍历node.expressions即 GROUP BY 后的字段列表对每个字段检查它是否在SELECT子句的Projection节点中存在或是否是聚合函数如COUNT(*),SUM(amount)。这里有个坑SELECT a, COUNT(*) FROM t GROUP BY a合法但SELECT a, b, COUNT(*) FROM t GROUP BY a不合法b未聚合也未分组sqlglot的Group节点不直接包含SELECT子句需先node.find(sqlglot.exp.Select)获取父 Select 节点再遍历其expressions。Where节点检查条件安全性。重点找In、Like、Between子句中的字面量Literal节点如果值是用户输入的字符串需确认已转义如LIKE %{user_input}%中的{user_input}是否做过%_转义更进一步可检查WHERE中是否引用了高基数字段如user_id做等值查询避免索引失效。Join节点检查 JOIN 条件完备性。遍历node.on子句即ON a.id b.id部分确认左右两边的Column节点都指向有效表且类型可比较如INTvsVARCHAR需告警。特别注意CROSS JOIN笛卡尔积无ON条件必须强制要求WHERE中有至少一个等值过滤否则直接拦截。提示sqlglot的find方法返回第一个匹配节点find_all返回所有。遍历Column时务必用find_all因为SELECT a.name, b.age, c.phone FROM ...会有多个Column。另外node.parent属性在sqlglot中默认为None需手动在遍历时传递父节点引用否则无法向上追溯上下文。3.2 正则表达式不是写得越长越好而是要“够用、易维护、可审计”正则层的目标是“快、准、稳”不是“全”。我见过最离谱的正则一个表达式写了 200 多字符号称能检测所有 SQL 注入变种结果上线后连SELECT * FROM users WHERE name OReilly带撇号的英文名都匹配失败因为正则里写了[^]*试图排除单引号却忘了转义。正则的本质是模式识别不是逻辑推理所以设计原则就三条第一按风险等级分组编写不追求单条通吃。我把正则分成三类P0 级立即拦截r;\s*(CREATE|ALTER|DROP|TRUNCATE|DELETE|UPDATE)\b、r\b(xp_|sp_)SQL Server 系统存储过程、r--\s*NOCHECK绕过约束注释P1 级告警复核rLIKE\s%[^]*\%模糊查询通配符、r\b(UNION|INTERSECT|EXCEPT)\sSELECT\b联合查询可能用于数据探测P2 级记录观察r\b(SELECT|INSERT|UPDATE|DELETE)\b统计高频操作类型不拦截。第二所有正则必须带re.IGNORECASE和re.DOTALL标志。SQL 里关键字大小写混用太常见select,SELECT,SeLeCt不忽略大小写等于废了一半而DOTALL让.能匹配换行符因为用户输入的 SQL 经常换行缩进rSELECT.*FROM在多行时会失效。第三正则必须可审计、可回溯。每条正则后面紧跟注释说明匹配场景、风险类型、历史案例。例如# P0: 检测分号分隔的危险语句案例2023-08-15 用户输入123; DROP TABLE users; 导致误删 r;\s*(CREATE|ALTER|DROP|TRUNCATE|DELETE|UPDATE)\b, # P1: 检测 UNION SELECT案例2024-02-03 渗透测试发现攻击者用 id1 UNION SELECT password FROM users 探测 r\b(UNION|INTERSECT|EXCEPT)\sSELECT\b,这样当某条正则误报时运维能立刻查到是谁写的、为什么写、上次修改时间而不是对着一串乱码抓瞎。注意正则永远要配合re.escape()处理动态拼接的字符串。比如要检查用户输入的表名是否在黑名单里不能写rFROM\s table_name r\b而要写rFROM\s re.escape(table_name) r\b否则table_name users; DROP TABLE logs就会变成rFROM\susers; DROP TABLE logs\b直接注入成功。这是无数安全漏洞的根源。4. 实操过程从零搭建一个可落地的 SQL 检查服务4.1 环境准备与依赖安装我们用 Python 3.9 构建服务核心依赖只有三个精简到极致sqlglot11.4.3AST 解析主力版本锁死因为新版本可能调整节点结构如 12.x 把Table节点改名为Schemaregex2023.10.3比内置re更强大支持\p{L}Unicode 字母等高级特性处理多语言字段名更稳pydantic2.5.2定义输入输出 Schema自动生成文档和校验避免手写if not sql:这种弱校验。安装命令一行搞定pip install sqlglot11.4.3 regex2023.10.3 pydantic2.5.2为什么不用pandas或sqlalchemy前者太重纯文本检查不需要 DataFrame后者是 ORM解析 SQL 是副业且对复杂嵌套查询支持不如sqlglot。我们追求的是“小而美”一个检查函数输入 SQL 字符串和数据库类型输出CheckResult对象干净利落。4.2 核心检查函数实现含完整代码下面这段代码是我在线上跑了 18 个月的核心检查函数已脱敏可直接复制使用import re import sqlglot from sqlglot import exp from typing import List, Dict, Optional, Tuple, Any from pydantic import BaseModel, Field class CheckResult(BaseModel): is_safe: bool Field(..., description是否通过检查) errors: List[str] Field(default_factorylist, description错误信息列表) warnings: List[str] Field(default_factorylist, description警告信息列表) ast_info: Dict[str, Any] Field(default_factorydict, descriptionAST 解析摘要) def check_sql(sql: str, dialect: str, schema: Optional[Dict[str, List[str]]] None) - CheckResult: 主 SQL 检查函数 :param sql: 待检查的 SQL 字符串 :param dialect: 目标数据库方言如 postgres, mysql, tsql :param schema: 可选数据库 schema 映射格式 {users: [id, name, email], orders: [id, user_id, amount]} :return: CheckResult 对象 result CheckResult(is_safeTrue) # 第一层正则快筛 p0_patterns [ (r;\s*(CREATE|ALTER|DROP|TRUNCATE|DELETE|UPDATE)\b, 检测分号分隔的危险语句), (r\b(xp_|sp_), 检测 SQL Server 系统存储过程调用), (r--\s*NOCHECK, 检测绕过约束的注释), (r\b(UNION\sALL\sSELECT|INTERSECT\sSELECT|EXCEPT\sSELECT)\b, 检测高风险联合查询), ] for pattern, desc in p0_patterns: if re.search(pattern, sql, re.IGNORECASE | re.DOTALL): result.is_safe False result.errors.append(fP0级风险{desc}匹配内容{re.search(pattern, sql, re.IGNORECASE | re.DOTALL).group(0)}) return result # 第二层AST 解析与语义检查 try: parsed sqlglot.parse_one(sql, readdialect) result.ast_info[type] type(parsed).__name__ result.ast_info[dialect] dialect except Exception as e: result.is_safe False result.errors.append(fSQL 语法错误{str(e)}) return result # 检查字段存在性需 schema if schema: _check_column_existence(parsed, schema, result) # 检查 GROUP BY 完整性 _check_group_by(parsed, result) # 检查 WHERE 条件安全性简单版找未转义的 LIKE _check_like_safety(sql, result) # 检查 JOIN 条件 _check_join_safety(parsed, result) # 第三层可选的执行前模拟此处仅示意逻辑 # if should_simulate(sql): # 根据 SQL 复杂度或用户权限决定 # plan get_explain_plan(sql, dialect) # 调用数据库 EXPLAIN # if has_full_scan(plan): # result.warnings.append(执行计划显示全表扫描建议添加索引) return result def _check_column_existence(node: exp.Expression, schema: Dict[str, List[str]], result: CheckResult): 检查所有 Column 节点的字段是否存在 for col in node.find_all(exp.Column): table_name col.table.lower() if col.table else col_name col.name.lower().strip(\) # 去掉引号 # 如果有表前缀查对应表 if table_name and table_name in schema: if col_name not in schema[table_name]: result.is_safe False result.errors.append(f字段 {col_name} 在表 {table_name} 中不存在。可用字段{schema[table_name]}) # 如果无表前缀查所有表宽松策略生产环境建议禁用 elif not table_name: found False for tbl, cols in schema.items(): if col_name in cols: found True break if not found: result.is_safe False result.errors.append(f字段 {col_name} 在所有已知表中均不存在) def _check_group_by(node: exp.Expression, result: CheckResult): 检查 GROUP BY 是否覆盖所有非聚合字段 select_node node.find(exp.Select) if not select_node: return group_node node.find(exp.Group) if not group_node or not group_node.expressions: return # 获取 SELECT 中的所有非聚合字段 non_agg_columns [] for expr in select_node.expressions: if isinstance(expr, exp.Column): non_agg_columns.append(expr) elif isinstance(expr, exp.AggFunc): # 聚合函数跳过 continue else: # 其他表达式如 Literal、Binary视为常量不需分组 pass # 获取 GROUP BY 字段 group_cols [] for expr in group_node.expressions: if isinstance(expr, exp.Column): group_cols.append(expr) # 检查每个非聚合字段是否在 GROUP BY 中 for col in non_agg_columns: if col not in group_cols: result.is_safe False result.errors.append(f字段 {col.name} 在 SELECT 中出现但未在 GROUP BY 中需聚合或分组) def _check_like_safety(sql: str, result: CheckResult): 检查 LIKE 子句中是否有未转义的通配符 # 简单匹配 LIKE xxx% 模式 like_matches re.findall(rLIKE\s([^]*), sql, re.IGNORECASE | re.DOTALL) for pattern in like_matches: if % in pattern or _ in pattern: # 检查是否已转义如 \% 或 \_ if not (r\% in pattern or r\_ in pattern): result.warnings.append(fLIKE 模式 {pattern} 包含未转义通配符可能导致全表扫描) def _check_join_safety(node: exp.Expression, result: CheckResult): 检查 JOIN 是否有 ON 条件或隐式笛卡尔积 for join in node.find_all(exp.Join): if not join.kind or join.kind.upper() CROSS: result.is_safe False result.errors.append(检测到 CROSS JOIN笛卡尔积必须提供 ON 条件或 WHERE 过滤) elif not join.on: result.is_safe False result.errors.append(JOIN 缺少 ON 条件可能导致笛卡尔积)这段代码的关键在于错误分级明确P0 错误直接return不继续检查避免无效计算schema 传参灵活schema是可选参数方便单元测试传 mock schema和生产环境传真实 schema cache字段名清洗严谨col.name.lower().strip(\)同时处理大小写和引号因为SELECT User_Name和SELECT User_Name是等价的GROUP BY 检查务实只检查Column类型字段忽略COUNT(*)等聚合符合实际业务逻辑。4.3 集成到 Text2SQL Pipeline 的实操步骤检查服务不是孤立的必须无缝嵌入现有流程。我们典型的 Text2SQL pipeline 是用户提问 → LLM 生成 SQL →SQL 检查服务→ 安全 SQL → 数据库执行 → 返回结果。集成只需三步第一步暴露 HTTP 接口用 FastAPI 包一层让检查服务变成 REST APIfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class CheckRequest(BaseModel): sql: str dialect: str schema: Optional[Dict[str, List[str]]] None app.post(/check) def check_sql_endpoint(req: CheckRequest): try: result check_sql(req.sql, req.dialect, req.schema) return result.model_dump() except Exception as e: raise HTTPException(status_code400, detailf检查失败{str(e)})启动命令uvicorn main:app --host 0.0.0.0:8000 --reload接口地址http://localhost:8000/check。第二步在 LLM 生成后调用检查假设你的 Text2SQL 模型输出是generated_sql在调用数据库前插入检查逻辑import requests def execute_text2sql(question: str, db_type: str) - dict: # Step 1: LLM 生成 SQL此处省略具体模型调用 generated_sql llm_generate(question, db_type) # Step 2: 调用检查服务 check_resp requests.post( http://localhost:8000/check, json{ sql: generated_sql, dialect: db_type, schema: get_current_schema(db_type) # 从缓存或 DB 获取实时 schema } ) if check_resp.status_code ! 200: raise RuntimeError(fSQL 检查服务异常{check_resp.text}) check_result check_resp.json() if not check_result[is_safe]: # 根据错误类型做不同处理 if any(P0级风险 in err for err in check_result[errors]): log_security_alert(check_result[errors]) # 记录安全事件 raise PermissionError(检测到高危 SQL已拦截) else: # 非 P0 错误可降级为重试或提示用户 return {status: warning, message: SQL 存在潜在问题, details: check_result} # Step 3: 安全 SQL 执行 return execute_raw_sql(check_result[sql], db_type)第三步schema 动态同步机制schema 不能硬编码必须随数据库变更自动更新。我们用两种方式定时同步每小时跑一次SELECT table_name, column_name FROM information_schema.columns WHERE table_schemapublicPostgreSQL或SELECT name FROM sys.columns WHERE object_id OBJECT_ID(table_name)SQL Server更新 Redis 缓存事件驱动监听数据库 DDL 事件如CREATE TABLE通过 CDC 工具Debezium捕获触发 schema 更新。线上用的是定时同步因为 DDL 操作频率低且事件驱动增加架构复杂度。实操心得第一次上线时我们把 schema 同步间隔设为 5 分钟结果遇到一个紧急需求DBA 手动加了字段但检查服务还没同步导致合法 SQL 被误判“字段不存在”。后来改成“缓存失效 降级兜底”当检查发现字段不存在时不直接报错而是尝试调用DESCRIBE table_name实时查询如果实时查询成功则更新缓存并重试检查。这样既保证了强一致性又避免了单点故障。5. 常见问题与排查技巧实录5.1 典型问题速查表问题现象可能原因排查步骤解决方案正则误报SELECT * FROM users WHERE name OReilly被拦截正则写了r[^]*试图匹配字符串字面量但OReilly中的两个单引号是 SQL 标准的转义写法正则未识别1. 用re.findall(r[^]*, sql)测试匹配结果2. 查看匹配到的字符串是否包含改用r.*?(?!)(?!)负向先行断言或直接放弃字符串字面量检查交给 AST 的Literal节点处理AST 检查漏报SELECT a.name, b.age FROM users a, orders b WHERE a.id b.user_id没报笛卡尔积这是旧式逗号 JOINsqlglot解析为Select下挂From和Where没有Join节点_check_join_safety函数未覆盖1.print(parsed)查看 AST 结构2. 检查parsed.find(exp.From)的joins属性是否为空在_check_join_safety前加逻辑if len(parsed.find(exp.From).expressions) 1:则视为隐式 JOIN强制要求WHERE中有AND连接的等值条件schema 同步延迟新表加了字段检查仍报“字段不存在”Redis 缓存未及时更新或get_current_schema()函数读取了过期缓存1.redis-cli GET schema:postgres查看缓存内容2. 手动执行SELECT * FROM information_schema.columns WHERE table_namenew_table验证 DB 状态在检查函数开头加if not schema: schema get_fallback_schema()fallback 逻辑是直连 DB 查询确保最终一致性性能瓶颈单次检查耗时超 500mssqlglot.parse_one()对超长 SQL5000 字符解析慢或正则用了贪婪匹配.*1.timeit测试parse_one和正则耗时2. 用re.compile()预编译所有正则对 SQL 截断sql[:2000]检查只关注结构不关心超长 VALUES所有正则用re.compile()预编译并加^$锚点限制匹配范围5.2 独家避坑技巧分享技巧一用“最小可运行 SQL”做单元测试而不是用完整业务 SQL别一上来就拿SELECT u.name, o.amount, p.product_name FROM users u JOIN orders o ON u.ido.user_id JOIN products p ON o.product_idp.id WHERE u.created_at 2024-01-01 GROUP BY u.name, o.amount, p.product_name ORDER BY o.amount DESC LIMIT 10这种复杂语句测试。先写最简 caseSELECT * FROM users基础语法SELECT name FROM users WHERE id 1WHERE 检查SELECT COUNT(*) FROM users GROUP BY statusGROUP BY 检查SELECT * FROM users; DROP TABLE logs正则拦截每个 case 对应一个assert确保修改代码不破坏已有逻辑。我们团队有 87 个这样的单元测试覆盖率 92%每次发版前 5 分钟跑完比人工回归高效十倍。技巧二对“合法但危险”的 SQL不拦截而是打标签 人工复核流比如SELECT * FROM large_table WHERE created_date 2020-01-01语法语义全对但large_table有 10 亿行created_date无索引。这种 SQL 不该被检查服务拒掉业务可能真需要但必须标记。我们在CheckResult里加了risk_level: str low|medium|high字段high级别如全表扫描、笛卡尔积走审批流medium如无索引 WHERE、大结果集 LIMIT发企业微信告警low如字段存在性警告只记日志。这样平衡了安全与效率。技巧三日志必须记录“原始 SQL 检查结果 时间戳 请求 ID”缺一不可有一次线上报错“SQL 检查失败”但日志只写了{is_safe: false, errors: [字段不存在]}根本不知道是哪条 SQL、哪个用户、什么时间触发的。后来我们强制所有日志加request_id从 HTTP Header 透传并在检查函数入口打logger.info(fSQL_CHECK_START: {request_id} | SQL_LEN: {len(sql)} | DIALECT: {dialect})出口打logger.info(fSQL_CHECK_END: {request_id} | RESULT: {result.is_safe} | ERRORS: {result.errors})。现在任何问题运维同学 30 秒内就能定位到原始请求再也不用翻三天日志。技巧四给业务方提供“检查报告”页面而不是只返回 success/fail前端加一个/sql-check-report页面用户粘贴 SQL点击检查返回可视化报告左侧显示原始 SQL高亮标出被拦截的关键词右侧分 Tab 展示“正则扫描结果”、“AST 语义分析”、“安全建议”。比如SELECT * FROM users WHERE name LIKE %{input}%报告会明确写“检测到未转义 LIKE 模式建议将{input}替换为escape_like({input})函数”。这比抛异常友好太多业务方能自己修而不是每次找研发。6. 最后一点个人体会我在做这个检查模块之前一直以为 Text2SQL 的难点在模型本身——怎么让大模型理解“环比增长”“同比下滑”这些业务术语。做完之后才明白真正的难点在模型之后**如何让机器生成的