智能SQL审核系统设计用LLM检测危险DDL并生成安全替代方案一、一个DROP TABLE引发的凌晨紧急回滚凌晨 2:15运维群里弹出一条消息有人在生产库执行了ALTER TABLE orders DROP COLUMN total_amount线上的订单金额全部变 NULL 了。回滚 Binlog、恢复备份、补数据、修复下游任务——整个团队忙到天亮。事后复盘发现这条 DDL 来自一个看起来无害的 JIRA 工单上的 SQL 脚本由一位初级工程师编写。问题不是工程师的水平而是缺少一个自动化的 DDL 审核门禁——如果这条 SQL 在到达数据库之前就被拦截并提示风险整个事故就不会发生。传统的 SQL 审核依赖规则引擎如 Yearning、Inception通过硬编码的规则表检测危险关键字DROP、TRUNCATE和索引缺失。但规则引擎有三个天生短板无法理解业务语义不知道total_amount是核心字段、无法评估影响范围不知道这个删除会影响多少下游、无法生成替代方案只能拒绝不能建议。LLM 的引入将 SQL 审核从模式匹配升级到语义理解——不仅能检测危险关键字还能理解表结构、评估影响、并生成安全的替代方案。二、多层审核架构从语法检测到语义评估flowchart TB A[DDL/DML SQL] -- B{第一层语法解析} B --|语法错误| R1[直接拒绝] B --|通过| C{第二层规则引擎} C --|命中危险规则| R2[高风险拦截] C --|通过| D{第三层LLM语义审核} subgraph LLM[LLM 审核层] D -- E[Schema 上下文注入] E -- F[风险评估] F -- G[影响范围分析] G -- H{风险等级} H --|高风险| I[生成安全替代方案] H --|中风险| J[人工审批 建议] H --|低风险| K[自动通过] end subgraph Feedback[反馈闭环] L[审批结果记录] -- M[规则学习] M -- C end三层防御策略第一层语法解析用 MySQL 解析器验证 SQL 语法正确性——这是零成本的校验第二层规则引擎基于启发式规则快速拦截已知危险模式DROP、TRUNCATE、ALTER ... CHANGE第三层LLM 语义审核这是本文的核心——注入完整的 Schema 上下文让 LLM 评估 SQL 的业务风险三层中规则引擎仍然是第一道防线因为它在 0ms 内完成LLM 审核作为深度语义检查的补充耗时 1-3 秒。三、完整系统实现3.1 Schema 上下文构建import pymysql from typing import Dict, List, Optional from dataclasses import dataclass import json dataclass class ColumnInfo: name: str data_type: str nullable: bool default_value: Optional[str] comment: str in_indexes: List[str] dataclass class TableContext: 包含完整的表结构信息用于 LLM 审核 table_name: str engine: str row_count_estimate: int columns: List[ColumnInfo] indexes: List[Dict] foreign_keys: List[Dict] dependent_views: List[str] recent_query_count: int # 近24小时查询次数 class SchemaContextFetcher: 提取目标表的完整上下文 def __init__(self, db_config: dict): self.conn pymysql.connect(**db_config) def fetch(self, table_name: str) - TableContext: 提取表的所有元信息 # 1. 基本表信息 table_info self._get_table_info(table_name) # 2. 列信息 索引关联 columns self._get_columns_with_index(table_name) # 3. 索引信息 indexes self._get_indexes(table_name) # 4. 外键与依赖关系 foreign_keys self._get_foreign_keys(table_name) dependent_views self._get_dependent_views(table_name) # 5. 近24小时查询频率来自 Performance Schema recent_queries self._get_recent_query_count(table_name) return TableContext( table_nametable_name, enginetable_info.get(ENGINE, ), row_count_estimatetable_info.get(TABLE_ROWS, 0), columnscolumns, indexesindexes, foreign_keysforeign_keys, dependent_viewsdependent_views, recent_query_countrecent_queries ) def _get_columns_with_index(self, table_name: str) - List[ColumnInfo]: 获取列信息并关联索引 cursor self.conn.cursor(pymysql.cursors.DictCursor) # 列信息 cursor.execute(fSHOW FULL COLUMNS FROM {table_name}) columns cursor.fetchall() # 索引信息哪些列在索引中 cursor.execute(fSHOW INDEX FROM {table_name}) index_rows cursor.fetchall() # 构建列名 → 索引列表的映射 col_indexes {} for row in index_rows: col_name row[Column_name] if col_name not in col_indexes: col_indexes[col_name] [] col_indexes[col_name].append(row[Key_name]) result [] for col in columns: result.append(ColumnInfo( namecol[Field], data_typecol[Type], nullablecol[Null] YES, default_valuecol.get(Default), commentcol.get(Comment, ), in_indexescol_indexes.get(col[Field], []) )) return result def _get_foreign_keys(self, table_name: str) - List[Dict]: 获取外键依赖关系 cursor self.conn.cursor(pymysql.cursors.DictCursor) cursor.execute( SELECT CONSTRAINT_NAME, COLUMN_NAME, REFERENCED_TABLE_NAME, REFERENCED_COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA DATABASE() AND TABLE_NAME %s AND REFERENCED_TABLE_NAME IS NOT NULL , (table_name,)) return cursor.fetchall()3.2 LLM SQL 审核引擎from typing import Tuple, Optional class SQLAuditor: 基于 LLM 的 SQL 语义审核器 AUDIT_PROMPT 你是一个 MySQL DBA 专家。请审核以下 SQL 语句评估其安全风险。 ## 目标表上下文 {table_context} ## SQL 语句 {sql_statement} ## 审核要求 请从以下维度评估风险并以 JSON 格式返回结果 1. **risk_level**: 风险等级 (HIGH / MEDIUM / LOW) 2. **primary_risk**: 主要风险描述如果安全则填无 3. **impact_analysis**: - 影响行数预估 - 影响的下游依赖视图、外键表 - 是否会影响业务核心流程 - 回滚难度评估 4. **safer_alternative**: 如果有风险提供更安全的替代方案 SQL如果安全则填空字符串 5. **pre_execution_steps**: 建议执行前的准备步骤如备份、从库验证 6. **rollback_plan**: 如果出现问题回滚方案 返回格式严格 JSON {{ risk_level: HIGH|MEDIUM|LOW, primary_risk: 描述, impact_analysis: {{ estimated_rows: 0, dependent_objects: [], business_critical: true, rollback_difficulty: EASY|MEDIUM|HARD|IMPOSSIBLE }}, safer_alternative: SQL 或空字符串, pre_execution_steps: [步骤1, 步骤2], rollback_plan: 回滚方案 }} def __init__(self, schema_fetcher: SchemaContextFetcher, llm_client): self.schema_fetcher schema_fetcher self.llm llm_client def audit(self, sql: str) - Dict: 审核 SQL 语句 # 1. 从 SQL 中提取目标表名 tables self._extract_table_names(sql) # 2. 获取所有目标表的 Schema 上下文 context_parts [] for table in tables: ctx self.schema_fetcher.fetch(table) context_parts.append(self._format_context(ctx)) full_context \n\n.join(context_parts) # 3. 调用 LLM 审核 prompt self.AUDIT_PROMPT.format( table_contextfull_context, sql_statementsql ) response self.llm.chat( system你是一个专业的 MySQL DBA 审核专家请严格按照 JSON 格式返回审核结果。, user_messageprompt, temperature0.1 ) # 4. 解析结果 result self._parse_response(response) return result def _format_context(self, ctx: TableContext) - str: 将表上下文格式化为 LLM 友好的文本 lines [ f表名: {ctx.table_name}, f引擎: {ctx.engine}, f预估行数: {ctx.row_count_estimate:,}, f近24小时查询次数: {ctx.recent_query_count}, , 列信息:, ] for col in ctx.columns: index_note f [索引: {, .join(col.in_indexes)}] if col.in_indexes else nullable NULL if col.nullable else NOT NULL lines.append(f - {col.name} ({col.data_type}, {nullable}){index_note} -- {col.comment}) if ctx.foreign_keys: lines.append(\n外键依赖:) for fk in ctx.foreign_keys: lines.append( f - {fk[COLUMN_NAME]} → {fk[REFERENCED_TABLE_NAME]}.{fk[REFERENCED_COLUMN_NAME]} ) if ctx.dependent_views: lines.append(f\n依赖视图: {, .join(ctx.dependent_views)}) return \n.join(lines) def _extract_table_names(self, sql: str) - List[str]: 从 SQL 中提取表名简化实现 import re patterns [ r(?:ALTER|DROP|TRUNCATE|RENAME)\sTABLE\s(?:IF\sEXISTS\s)??(\w)?, r(?:FROM|JOIN|UPDATE|INTO)\s?(\w)?, ] tables [] for pattern in patterns: matches re.findall(pattern, sql, re.IGNORECASE) tables.extend(matches) return list(set(tables)) def _parse_response(self, response: str) - Dict: 解析 LLM 返回的 JSON import re # 提取 JSON 块 json_match re.search(rjson\s*(.*?), response, re.DOTALL) if json_match: response json_match.group(1) try: return json.loads(response) except json.JSONDecodeError: # 尝试修复常见错误 # 移除可能的注释和尾随逗号 cleaned re.sub(r//.*, , response) cleaned re.sub(r,\s*}, }, cleaned) cleaned re.sub(r,\s*], ], cleaned) return json.loads(cleaned)四、边界与工程代价审核准确性 vs 延迟LLM 审核增加 1-3 秒的延迟对于工单审批流程完全可以接受但对于自动化 CI/CD pipeline 可能成为瓶颈。实践中采用异步审核模式低风险 SQL 先放行高风险 SQL 必须等待 LLM 审核结果。Schema 变更窗口的一致性问题DDL 执行后缓存的 Schema 立即过期。如果下一分钟有人提交了另一个 SQL 审核使用过期 Schema 的审核结果可能不准确。需要设置缓存 TTL如 30 秒 主动失效机制。LLM 幻觉风险在极端复杂的多表 JOIN DDL 场景中LLM 可能幻觉出不存在的外键关系。因此LLM 审核的结果只能是建议规则引擎的拦截结果才是决策。五、总结智能 SQL 审核系统的核心价值在于将 DBA 经验转化为可自动化执行的审核规则三层防御互补语法层0ms→ 规则层1ms→ LLM 语义层1-3s各层独立决策LLM 不是替代规则而是增强规则对于规则引擎覆盖不到的场景如删除核心字段、LLM 提供的语义理解和替代方案生成是独一无二的价值审核结果必须可追溯每一条 SQL 的审核记录审核者、风险等级、替代建议都需要持久化这是合规和复盘的基础在实际部署中这套系统在上线第一个月就拦截了 17 条高危 DDL其中 4 条如果直接执行会造成生产事故生成了 23 条安全替代方案建议。拦截准确率 100%无误报导致正常 SQL 被拒绝是对 DBA 神经紧张的最好解药。