相信不少后端同学都有过这样的经历辛辛苦苦写一个统计 SQL在 JOIN、子查询、窗口函数之间反复横跳最后还是得靠一段段打印日志来排查问题。于是你开始尝试让 AI 写 SQL把表结构丢给它它能在几秒钟内生成一条像模像样的查询语句。可问题也随之而来——你真的敢让 AI 直接连接数据库去执行吗这篇文章要讨论的不是“用 AI 生成 SQL 语句”而是“如何安全地把数据库查询交给 AI 执行”。我们会一起搭建一个AI 查询网关AI Query Gateway让用户用自然语言提问AI 负责生成 SQL而校验和执行则由我们自己掌控。这样做的好处是既能享受 AI 带来的效率提升又能避免 AI 误操作、越权访问、拖垮数据库等问题。无论你是后端开发、数据分析工程师还是在做智能客服、报表问答、内部数据助手这套方案都有很强的参考价值。下面从风险分析、环境准备、核心设计、完整代码实现、常见排错到工程化建议一次讲透。1. 为什么“把数据库查询交给 AI”让人不放心1.1 AI 生成 SQL 的现状现在的 AI 大模型在 SQL 生成方面已经相当成熟。只要给它一份清晰的表结构说明它就能写出多表关联、聚合统计、窗口函数等复杂查询。对很多开发者来说这已经从“玩具”变成了“生产力工具”。举个典型场景运营人员想知道“上个季度每个城市的订单总额和订单数”。如果用传统方式你得先了解表结构再写一条带 JOIN 和 GROUP BY 的 SQL最后可能还要调整字段格式。如果用 AI直接把问题丢过去它可能几秒钟就返回一条 SQLSELECT c.city, COUNT(o.id) AS order_count, SUM(o.total_amount) AS total_amount FROM orders o JOIN customers c ON o.customer_id c.id WHERE o.created_at 2024-01-01 AND o.created_at 2024-04-01 GROUP BY c.city ORDER BY total_amount DESC;这条 SQL 大概率是正确的。但问题在于你愿意让这个 SQL 直接在公司的生产数据库上跑吗如果愿意你的底气来自哪里是 AI 不会犯错还是你的系统架构已经能兜住它的错误答案是绝大多数情况下我们缺少的是后者。1.2 直接交给 AI 执行的危险点把 AI 生成的 SQL 直接丢到数据库执行会遇到几类典型风险。第一误操作风险。AI 的核心能力是“模仿和生成”它并不真正理解你的业务约束。如果 Prompt 写得不够严格或者表结构信息不完整它可能生成 DELETE、UPDATE、DROP 等危险语句。一旦执行后果很难挽回。第二越权访问风险。如果服务账号对数据库拥有全部权限AI 生成的 SQL 就能读取它本不该读取的数据。比如你只想开放订单表但数据库账号有权限访问用户表、支付表AI 也可能在上下文不明确时把敏感数据查出来。第三性能风险。自然语言描述可能很模糊AI 可能生成不带 WHERE 条件的全表扫描或对巨大数据表做笛卡尔积连接。一个慢查询就能让数据库 CPU 飙高影响整个业务链路。第四不可审计。没有统一的入口用户用 AI 查询过什么、生成的 SQL 是什么、执行时间有多长全部不可追踪。出了问题很难回溯。第五SQL 注入与恶意输入。如果 AI 服务缺少输入校验用户可能在问题里注入恶意指令诱导 AI 生成危险 SQL。不要以为这是危言耸听Prompt 注入在真实业务里已经屡见不鲜。1.3 解决思路AI 查询网关把这些风险摊开来看你就会发现问题不在于 AI 能不能写 SQL而在于我们能不能控制 AI 写出来的 SQL 怎么执行。所以需要引入一个AI 查询网关作为用户与数据库之间的中间层。整个流程变成用户自然语言提问 ↓ AI 查询网关 ├── 1. 调用大模型生成 SQL ├── 2. 对 SQL 做安全校验 ├── 3. 加上执行限制超时、行数 └── 4. 使用只读数据库账号执行 ↓ 返回结构化查询结果AI 只负责“生成”不负责“执行”。校验、限制、审计都放在网关里由我们自己的代码控制。这样即使 AI 生成了一条危险 SQL它也过不了校验这一关即使它生成了一条正确的但性能很差的 SQL执行层也能通过超时和行数限制把它兜住。2. 环境准备与方案选型2.1 技术栈说明本文的实战示例以 Python 为主涉及以下组件Python 3.9运行环境。FastAPI提供 HTTP 接口接收自然语言问题并返回查询结果。SQLAlchemy统一管理数据库连接方便切换不同类型的数据库。PyMySQLMySQL 驱动本文以 MySQL 为例。openai 或其他模型 SDK调用大模型生成 SQL具体以你实际使用的模型服务为准。uvicorn启动 FastAPI 服务。版本需要根据你的项目实际情况调整。本文示例以常见环境为例重点演示实现思路而不是绑定某个特定版本。2.2 数据库账号准备这是整个方案里最容易忽略、又最重要的一步给 AI 查询服务单独创建一个数据库账号并且只授予 SELECT 权限。以 MySQL 为例可以在测试库中创建这样的账号-- 只读账号只能查询 shop 库 CREATE USER ai_reader% IDENTIFIED BY 请换成强密码; GRANT SELECT ON shop.* TO ai_reader%; FLUSH PRIVILEGES;在开发环境这样做可能显得有点麻烦。但请相信我一旦进入生产环境这个只读账号能帮你拦住大量事故。绝对不要让 AI 服务使用具有 INSERT、UPDATE、DELETE 权限的账号更不要使用 root。2.3 项目结构规划为了让代码清晰可维护我建议把不同职责拆到不同文件里ai_query_gateway/ ├── requirements.txt ├── database.py # 数据库连接管理 ├── sql_validator.py # SQL 安全校验 ├── ai_client.py # 大模型调用封装 ├── main.py # FastAPI 接口 └── README.md后面我们会逐个文件实现。先看一下整体依赖# requirements.txt fastapi uvicorn sqlalchemy pymysql openai以上依赖以你实际环境可用的版本为准。安装命令pip install -r requirements.txt3. 核心设计AI 查询网关的三层防护3.1 录入层自然语言转 SQL录入层的职责是把用户输入的自然语言问题转成一条结构化查询语句。这里最关键的是 Prompt 设计。你不能只说“帮我把订单查一下”而是要在 Prompt 里明确告诉 AI可用的表有哪些每张表有哪些字段。字段类型、含义以及关联关系。只能生成 SELECT 查询。禁止生成 INSERT、UPDATE、DELETE、DROP 等语句。默认返回行数限制。只输出 SQL不要输出解释。一个典型的系统 Prompt 可以这样设计你是数据库查询助手。你只能根据提供的表结构生成 SELECT 查询。 要求 1. 只允许 SELECT 或 WITH 开头。 2. 只能使用提供的表和字段。 3. 不允许出现 INSERT、UPDATE、DELETE、DROP、ALTER、TRUNCATE、GRANT 等关键字。 4. 如果问题无法用现有表结构回答请直接说明。 5. 默认限制返回不超过 200 行。 6. 只输出 SQL不要输出任何解释。这个 Prompt 看起来简单但它决定了 AI 输出的边界。很多 AI 生成错误 SQL 的情况其实不是模型能力不够而是 Prompt 没有给足约束。3.2 校验层白名单与 SQL 语义检查AI 返回的 SQL在真正执行之前必须经过校验层。这一层要做到两件事先拦危险语句再做语义检查。第一步做语法级检查。把 SQL 变成小写后判断第一个关键字是不是select或with。这一步能拦截大多数 DDL、DML 语句。第二步做关键词黑名单。即使语句以 SELECT 开头也可能包含子查询、注释、联合查询等绕过手段。所以要检查是否出现insert、update、delete、drop、truncate、alter、grant、exec、copy等危险关键字。第三步去掉注释。有些恶意 SQL 会用注释符把后面的校验代码注释掉。因此在校验前要先把--注释和/* */注释剥离。第四步检查分号。一条查询语句末尾可以有分号但不能在中间出现多个分号。更稳妥的做法是校验后直接去掉末尾分号避免一次提交多条语句。第五步追加 LIMIT。在 SQL 末尾自动加上LIMIT 条数。如果 AI 生成的 SQL 已经包含 LIMIT则尽量保留原样同时检查 LIMIT 数值是否超过上限。这个步骤能有效避免一次性拉回全表数据。3.3 执行层只读连接、超时与行数限制即使 SQL 通过了校验执行时也要防一手。只读连接。数据库连接使用前面创建的低权限只读账号。这里要注意不要让连接串使用业务系统的高权限账号。超时控制。SQL 执行要有超时时间。MySQL 可以在会话级别设置max_execution_timeSQLAlchemy 连接层面也可以设置超时。对于慢查询宁可让它超时失败也不要让它占用数据库资源。结果行数限制。在代码里使用fetchmany(n)而不是fetchall()。即使 SQL 本身没有 LIMIT代码层也不能把所有数据一次性读进内存。这三层设计做完AI 查询网关的骨架已经出来了。下面进入完整实战。4. 完整实战搭建一个可运行的 AI 查询网关4.1 创建项目与依赖新建项目目录并在其中创建虚拟环境mkdir ai_query_gateway cd ai_query_gateway python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate安装依赖pip install fastapi uvicorn sqlalchemy pymysql openai4.2 准备测试数据本文以 MySQL 为例假设有一个业务库shop包含客户表和订单表。在测试环境执行下面的建表语句USE shop; CREATE TABLE customers ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL, city VARCHAR(50) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE orders ( id INT PRIMARY KEY AUTO_INCREMENT, customer_id INT NOT NULL, total_amount DECIMAL(10,2) NOT NULL, status VARCHAR(20) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_customer (customer_id), KEY idx_created (created_at) );插入几条样例数据INSERT INTO customers (name, city) VALUES (张三, 上海), (李四, 北京), (王五, 上海); INSERT INTO orders (customer_id, total_amount, status, created_at) VALUES (1, 100.00, 已完成, 2024-03-01 10:00:00), (2, 250.50, 已完成, 2024-03-05 14:30:00), (3, 80.00, 待支付, 2024-04-02 09:00:00), (1, 320.00, 已完成, 2024-04-10 20:00:00);这些数据只是用来验证程序可以跑通你可以换成自己的业务表。4.3 实现数据库连接管理文件路径database.pyimport os from sqlalchemy import create_engine # 请根据实际环境修改连接串建议使用只读账号 DATABASE_URL os.getenv( DATABASE_URL, mysqlpymysql://ai_reader:你的密码127.0.0.1:3306/shop, ) engine create_engine( DATABASE_URL, pool_pre_pingTrue, pool_recycle3600, )这里需要注意两点连接串中的账号是ai_reader这是一个只读账号不是业务主账号。pool_pre_ping会在每次取连接前先探测连接是否可用避免拿到失效连接。pool_recycle设置连接回收时间防止 MySQL 的wait_timeout断掉长连接。4.4 实现 SQL 安全校验器文件路径sql_validator.py这是整个网关中最重要的一个文件。它的核心职责是“不信任 AI 输出的任何内容”。import re SELECT_KEYWORDS {select, with} FORBIDDEN_KEYWORDS [ insert, update, delete, replace, drop, truncate, alter, create, grant, revoke, exec, execute, call, copy, merge, vacuum, ] DEFAULT_LIMIT 100 MAX_LIMIT 200 def normalize_sql(sql: str) - str: 去掉注释、字符串内容和末尾分号返回统一格式的 SQL。 sql re.sub(r--[^\n]*, , sql) sql re.sub(r/\*.*?\*/, , sql, flagsre.S) sql re.sub(r[^]*, , sql) sql sql.strip().strip(;).strip() return sql def extract_limit(sql: str) - int | None: 从 SQL 中提取 LIMIT 数值没有则返回 None。 match re.search(r\blimit\s(\d), sql, flagsre.IGNORECASE) if match: return int(match.group(1)) return None def enforce_limit(sql: str, max_rows: int MAX_LIMIT) - str: 如果 SQL 没有 LIMIT就自动追加 LIMIT。 sql sql.strip().rstrip(;).strip() if extract_limit(sql) is None: return f{sql} LIMIT {max_rows} return sql def validate_select_only(sql: str) - bool: 校验 SQL 是否为只读查询返回 True 表示通过。 normalized normalize_sql(sql) if not normalized: return False first_keyword normalized.split()[0].lower() if first_keyword not in SELECT_KEYWORDS: return False lower_sql normalized.lower() for keyword in FORBIDDEN_KEYWORDS: if re.search(rf\b{keyword}\b, lower_sql): return False # 检查是否一次提交了多条语句 if ; in normalized: return False limit extract_limit(normalized) if limit is not None and limit MAX_LIMIT: return False return True这里的关键点在于先剥离注释防止 AI 在 SQL 里加注释干扰判断。把字符串字面量替换为空避免字符串内容里夹带危险关键词。只允许select或with开头从语法层面拦截危险操作。对分号做严格检查避免堆叠多条语句。4.5 实现大模型调用层文件路径ai_client.py大模型接口每家服务商都不一样本文以 OpenAI 兼容接口为例这是目前很多大模型服务都支持的调用方式。如果你的模型服务不兼容这个格式只需要把call_llm_to_generate_sql内部实现替换成你自己的 SDK 即可。import os import re from openai import OpenAI client OpenAI( api_keyos.getenv(LLM_API_KEY, 你的 API Key), base_urlos.getenv(LLM_BASE_URL, 你的模型服务地址), ) SYSTEM_PROMPT 你是数据库查询助手。你只能根据提供的表结构生成 SELECT 查询。 要求 1. 只能使用 SELECT 或 WITH 开头。 2. 只能使用提供的表和字段。 3. 不允许出现 INSERT、UPDATE、DELETE、DROP、ALTER、TRUNCATE、GRANT 等关键字。 4. 如果问题无法用现有表结构回答请直接说明。 5. 默认限制返回不超过 200 行。 6. 只输出 SQL不要输出任何解释。 TABLE_SCHEMA 表 customers - id INT 主键 - name VARCHAR 客户姓名 - city VARCHAR 城市 - created_at DATETIME 创建时间 表 orders - id INT 主键 - customer_id INT 关联 customers.id - total_amount DECIMAL 订单金额 - status VARCHAR 订单状态 - created_at DATETIME 创建时间 def call_llm_to_generate_sql(question: str) - str: response client.chat.completions.create( modelos.getenv(LLM_MODEL, 你的模型名称), temperature0.0, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: f表结构如下\n{TABLE_SCHEMA}\n\n用户问题{question}}, ], ) sql response.choices[0].message.content.strip() # 如果模型把 SQL 放在代码块中去掉多余标记 sql re.sub(r^(sql)?, , sql).strip() sql re.sub(r$, , sql).strip() return sql这段代码需要根据你的大模型服务实际情况调整。没有标准答案关键是把握两个原则系统 Prompt 一定要写清楚“只允许 SELECT”。模型输出后要做基础清洗去掉代码块标记。4.6 实现 FastAPI 查询接口文件路径main.pyimport time from fastapi import FastAPI, HTTPException from pydantic import BaseModel from sqlalchemy import text from ai_client import call_llm_to_generate_sql from database import engine from sql_validator import ( enforce_limit, validate_select_only, DEFAULT_LIMIT, ) app FastAPI(titleAI 数据查询网关) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): sql: str columns: list rows: list row_count: int cost_ms: int app.post(/query, response_modelQueryResponse) def query_database(req: QueryRequest): # 1. 调用大模型生成 SQL sql call_llm_to_generate_sql(req.question) # 2. 安全校验失败直接拒绝 if not validate_select_only(sql): raise HTTPException(status_code400, detailAI 生成的 SQL 未通过安全校验已拒绝执行) # 3. 自动补 LIMIT sql enforce_limit(sql, max_rows200) # 4. 使用只读连接执行查询 start time.time() try: with engine.connect() as conn: result conn.execute(text(sql)) columns list(result.keys()) rows [list(row) for row in result.fetchmany(DEFAULT_LIMIT)] except Exception as exc: raise HTTPException(status_code500, detailf数据库执行失败{str(exc)}) cost_ms int((time.time() - start) * 1000) return QueryResponse( sqlsql, columnscolumns, rowsrows, row_countlen(rows), cost_mscost_ms, )这段代码的逻辑很清晰先让 AI 生成 SQL。校验 SQL 是否安全。自动补充 LIMIT。用只读连接执行最多取 100 行。返回 SQL、表头、数据行和执行耗时。4.7 启动服务并验证在项目根目录启动 FastAPI 服务uvicorn main:app --reload --port 8000新开一个终端用 curl 测试curl -X POST http://127.0.0.1:8000/query \ -H Content-Type: application/json \ -d {question: 上海客户一共下了多少订单}预期会返回类似下面的结构{ sql: SELECT COUNT(o.id) AS order_count FROM orders o JOIN customers c ON o.customer_id c.id WHERE c.city 上海;, columns: [order_count], rows: [[2]], row_count: 1, cost_ms: 35 }你也可以在浏览器打开http://127.0.0.1:8000/docs通过 Swagger UI 直接调试接口。为了验证安全校验是否生效可以故意构造一个危险的提问比如“把 orders 表删掉”。此时 AI 生成的 SQL 应该是 DELETE 或 DROP 语句校验器会返回 400 错误不会真正执行。5. 常见问题与排查思路5.1 AI 生成的 SQL 不对怎么办这是遇到最多的一个问题。AI 生成的 SQL 本身合法但查出来的结果和预期不一致。通常原因有几种表结构信息不完整字段含义描述不到位。用户问题存在歧义比如“最近一个月”没有给出明确基准日期。模型没有理解字段之间的关联关系。解决思路把表结构描述写得更加详细尤其是字段注释和枚举值。对于常见问题可以在 Prompt 里给出几个示例问答。如果业务复杂建议在网关前增加一层“语义层”提前把术语映射成具体 SQL 片段而不是让 AI 每次重新理解。5.2 查询太慢拖垮数据库怎么办即使加了 LIMIT一个没有 WHERE 条件的大表查询依然可能很慢。解决思路在数据库账号层面设置最大执行时间。MySQL 可以在连接建立后执行SET max_execution_time 5000。在 SQLAlchemy 连接上设置超时参数。在 SQL 校验器里检查 WHERE 条件如果一条查询没有 WHERE直接拒绝或要求用户确认。为常用维度的字段建索引。索引能显著提升查询速度。把 AI 查询路由到只读从库避免影响主库业务。5.3 用户问的问题超出表结构范围怎么办比如用户问“哪个客户信用分最高”但你的表里根本没有信用分字段。这种情况下AI 可能硬编造一个字段名导致 SQL 执行报错。更合理的处理方式在 Prompt 里明确要求 AI 在字段不存在时直接说明不要强行生成 SQL。网关收到这样的响应后返回一个友好的提示而不是把报错堆给用户。5.4 权限和审计怎么做权限方面的第一原则是AI 服务使用的数据库账号权限必须小于等于业务所需的最小权限。只读账号、最小权限、网络隔离缺一不可。审计方面建议在网关里记录以下信息用户问题原文。AI 生成的 SQL。校验结果。执行耗时。返回行数。用户身份和 IP。这些日志不需要太复杂保留到日志系统即可。将来出问题可以快速回溯是哪条 SQL、哪个用户、什么时间执行的。5.5 常见问题速查问题现象常见原因解决思路AI 生成 SQL 中包含不存在字段表结构描述不完整补充字段注释和枚举值必要时增加语义层危险 SQL 未被拦截校验规则不够严格完善正则和关键字黑名单或改用 AST 解析返回数据量过大没有 LIMIT 限制自动追加 LIMIT并使用 fetchmany查询执行很慢大表全表扫描建立索引、设置超时、限制 WHERE 必须存在用户绕过网关直接连库数据库账号权限过大使用只读账号限制来源 IP模型输出带解释文字Prompt 约束不足在 Prompt 中强调只输出 SQL并对输出做清洗6. 工程化最佳实践6.1 最小权限原则在数据库层面AI 查询服务应该使用独立账号密码不能写死在代码里而是通过环境变量或配置中心注入。账号权限只保留 SELECT甚至可以只授权特定的几张表或者视图。这里要特别提一个读者经常问的问题视图可以加快查询速度吗视图本质是一段保存好的查询逻辑它不会自动提升性能但它可以隐藏底层表结构统一业务口径。在 AI 查询场景中给 AI 暴露设计良好的视图比直接暴露原始表更安全也更不容易生成错误 SQL。所以不要指望视图能优化性能但可以把它当成一组“可控的查询入口”。6.2 尽可能使用 AST 解析而不是纯文本校验纯正则做 SQL 校验有很多边界问题比如 WHERE 条件里的字符串可能包含危险关键词。更稳妥的方案是使用 SQL 解析器例如sqlglot或sqlparse把 SQL 解析成语法树后判断节点类型。示例思路如下需要按实际版本调整import sqlglot from sqlglot import exp from sqlglot.errors import ParseError def validate_sql_by_ast(sql: str) - bool: try: expressions sqlglot.parse(sql) except ParseError: return False for expression in expressions: # SQLGlot 会把非 SELECT 表达式的类型暴露出来 if not isinstance(expression, exp.Select): return False # 继续检查是否包含 DDL / DML 节点 for node in expression.walk(): if isinstance(node, (exp.Insert, exp.Update, exp.Delete, exp.Drop)): return False return True用 AST 的好处是能真正理解 SQL 结构而不是靠字符串匹配猜。大型项目里强烈建议把校验层从正则升级成解析器。6.3 审计与监控网关是天然的统一入口。每次查询都应该记录日志建议包含以下字段timestamp: 2024-05-10 10:00:00 user_id: u_1001 question: 上海客户一共下了多少订单 generated_sql: SELECT ... validated: true execution_ms: 35 row_count: 1如果条件允许可以把这些日志同步到日志中心或数据库用于后续分析。监控方面重点关注失败率、平均耗时、被拦截的 SQL 数量。被拦截的次数突然上升往往说明 Prompt 需要调整或者用户开始尝试输入危险指令。6.4 Prompt 工程设计要持续迭代很多人把 Prompt 当成一次写完就固定下来的配置这是不对的。Prompt 需要根据真实问题持续迭代。一个建议的做法是先在测试库上跑一批典型问题把 AI 生成的错误 SQL 收集起来分析是表结构描述不清楚还是约束条件不明确再针对性地修改 Prompt。对于特别容易出错的问题可以在 Prompt 末尾追加“常见问题示例”。这个迭代过程比频繁更换大模型更有效。6.5 向量检索和语义缓存的扩展思路当你的业务表结构很多、常见的查询问题重复出现时可以考虑引入向量数据库。把“自然语言问题”和“正确 SQL”成对存入向量库收到新问题时先用向量检索召回相似的 SQL再让大模型基于召回结果做修改。这样做有两个好处一是降低大模型生成错误 SQL 的概率二是减少大模型调用成本。对高频问题可以完全走缓存不需要每次都调用模型。7. 总结与下一步学习方向这套 AI 查询网关的完整链路核心只有三句话AI 只负责生成校验必须独立执行必须受限。数据库权限用只读账号SQL 校验做白名单和关键词拦截执行时加超时和行数限制再把每一步都记录下来。有了这层网关你才真正可以放心地把数据库查询交给 AI。下一步建议你重点深入研究几个方向SQL 语法树解析把校验层从正则升级到 AST这是生产级方案的必经之路。语义层的构建用视图或中间表统一业务口径让 AI 面对的不再是一堆裸表。模型选型与成本控制选择适合 SQL 生成的模型评估准确率、延迟和调用成本。生产环境的隔离策略如果条件允许把 AI 查询指向只读从库与主库完全隔离。先把测试库跑通再逐步放开权限和场景。AI 查询这条路并不神秘但要靠一层一层的约束才能让它在真实业务中安全落地。如果这篇文章对你有帮助可以收藏备用后续有更好的工程实践我会继续补充。 AI查询网关实战：让AI生成SQL，安全执行由我掌控
相信不少后端同学都有过这样的经历辛辛苦苦写一个统计 SQL在 JOIN、子查询、窗口函数之间反复横跳最后还是得靠一段段打印日志来排查问题。于是你开始尝试让 AI 写 SQL把表结构丢给它它能在几秒钟内生成一条像模像样的查询语句。可问题也随之而来——你真的敢让 AI 直接连接数据库去执行吗这篇文章要讨论的不是“用 AI 生成 SQL 语句”而是“如何安全地把数据库查询交给 AI 执行”。我们会一起搭建一个AI 查询网关AI Query Gateway让用户用自然语言提问AI 负责生成 SQL而校验和执行则由我们自己掌控。这样做的好处是既能享受 AI 带来的效率提升又能避免 AI 误操作、越权访问、拖垮数据库等问题。无论你是后端开发、数据分析工程师还是在做智能客服、报表问答、内部数据助手这套方案都有很强的参考价值。下面从风险分析、环境准备、核心设计、完整代码实现、常见排错到工程化建议一次讲透。1. 为什么“把数据库查询交给 AI”让人不放心1.1 AI 生成 SQL 的现状现在的 AI 大模型在 SQL 生成方面已经相当成熟。只要给它一份清晰的表结构说明它就能写出多表关联、聚合统计、窗口函数等复杂查询。对很多开发者来说这已经从“玩具”变成了“生产力工具”。举个典型场景运营人员想知道“上个季度每个城市的订单总额和订单数”。如果用传统方式你得先了解表结构再写一条带 JOIN 和 GROUP BY 的 SQL最后可能还要调整字段格式。如果用 AI直接把问题丢过去它可能几秒钟就返回一条 SQLSELECT c.city, COUNT(o.id) AS order_count, SUM(o.total_amount) AS total_amount FROM orders o JOIN customers c ON o.customer_id c.id WHERE o.created_at 2024-01-01 AND o.created_at 2024-04-01 GROUP BY c.city ORDER BY total_amount DESC;这条 SQL 大概率是正确的。但问题在于你愿意让这个 SQL 直接在公司的生产数据库上跑吗如果愿意你的底气来自哪里是 AI 不会犯错还是你的系统架构已经能兜住它的错误答案是绝大多数情况下我们缺少的是后者。1.2 直接交给 AI 执行的危险点把 AI 生成的 SQL 直接丢到数据库执行会遇到几类典型风险。第一误操作风险。AI 的核心能力是“模仿和生成”它并不真正理解你的业务约束。如果 Prompt 写得不够严格或者表结构信息不完整它可能生成 DELETE、UPDATE、DROP 等危险语句。一旦执行后果很难挽回。第二越权访问风险。如果服务账号对数据库拥有全部权限AI 生成的 SQL 就能读取它本不该读取的数据。比如你只想开放订单表但数据库账号有权限访问用户表、支付表AI 也可能在上下文不明确时把敏感数据查出来。第三性能风险。自然语言描述可能很模糊AI 可能生成不带 WHERE 条件的全表扫描或对巨大数据表做笛卡尔积连接。一个慢查询就能让数据库 CPU 飙高影响整个业务链路。第四不可审计。没有统一的入口用户用 AI 查询过什么、生成的 SQL 是什么、执行时间有多长全部不可追踪。出了问题很难回溯。第五SQL 注入与恶意输入。如果 AI 服务缺少输入校验用户可能在问题里注入恶意指令诱导 AI 生成危险 SQL。不要以为这是危言耸听Prompt 注入在真实业务里已经屡见不鲜。1.3 解决思路AI 查询网关把这些风险摊开来看你就会发现问题不在于 AI 能不能写 SQL而在于我们能不能控制 AI 写出来的 SQL 怎么执行。所以需要引入一个AI 查询网关作为用户与数据库之间的中间层。整个流程变成用户自然语言提问 ↓ AI 查询网关 ├── 1. 调用大模型生成 SQL ├── 2. 对 SQL 做安全校验 ├── 3. 加上执行限制超时、行数 └── 4. 使用只读数据库账号执行 ↓ 返回结构化查询结果AI 只负责“生成”不负责“执行”。校验、限制、审计都放在网关里由我们自己的代码控制。这样即使 AI 生成了一条危险 SQL它也过不了校验这一关即使它生成了一条正确的但性能很差的 SQL执行层也能通过超时和行数限制把它兜住。2. 环境准备与方案选型2.1 技术栈说明本文的实战示例以 Python 为主涉及以下组件Python 3.9运行环境。FastAPI提供 HTTP 接口接收自然语言问题并返回查询结果。SQLAlchemy统一管理数据库连接方便切换不同类型的数据库。PyMySQLMySQL 驱动本文以 MySQL 为例。openai 或其他模型 SDK调用大模型生成 SQL具体以你实际使用的模型服务为准。uvicorn启动 FastAPI 服务。版本需要根据你的项目实际情况调整。本文示例以常见环境为例重点演示实现思路而不是绑定某个特定版本。2.2 数据库账号准备这是整个方案里最容易忽略、又最重要的一步给 AI 查询服务单独创建一个数据库账号并且只授予 SELECT 权限。以 MySQL 为例可以在测试库中创建这样的账号-- 只读账号只能查询 shop 库 CREATE USER ai_reader% IDENTIFIED BY 请换成强密码; GRANT SELECT ON shop.* TO ai_reader%; FLUSH PRIVILEGES;在开发环境这样做可能显得有点麻烦。但请相信我一旦进入生产环境这个只读账号能帮你拦住大量事故。绝对不要让 AI 服务使用具有 INSERT、UPDATE、DELETE 权限的账号更不要使用 root。2.3 项目结构规划为了让代码清晰可维护我建议把不同职责拆到不同文件里ai_query_gateway/ ├── requirements.txt ├── database.py # 数据库连接管理 ├── sql_validator.py # SQL 安全校验 ├── ai_client.py # 大模型调用封装 ├── main.py # FastAPI 接口 └── README.md后面我们会逐个文件实现。先看一下整体依赖# requirements.txt fastapi uvicorn sqlalchemy pymysql openai以上依赖以你实际环境可用的版本为准。安装命令pip install -r requirements.txt3. 核心设计AI 查询网关的三层防护3.1 录入层自然语言转 SQL录入层的职责是把用户输入的自然语言问题转成一条结构化查询语句。这里最关键的是 Prompt 设计。你不能只说“帮我把订单查一下”而是要在 Prompt 里明确告诉 AI可用的表有哪些每张表有哪些字段。字段类型、含义以及关联关系。只能生成 SELECT 查询。禁止生成 INSERT、UPDATE、DELETE、DROP 等语句。默认返回行数限制。只输出 SQL不要输出解释。一个典型的系统 Prompt 可以这样设计你是数据库查询助手。你只能根据提供的表结构生成 SELECT 查询。 要求 1. 只允许 SELECT 或 WITH 开头。 2. 只能使用提供的表和字段。 3. 不允许出现 INSERT、UPDATE、DELETE、DROP、ALTER、TRUNCATE、GRANT 等关键字。 4. 如果问题无法用现有表结构回答请直接说明。 5. 默认限制返回不超过 200 行。 6. 只输出 SQL不要输出任何解释。这个 Prompt 看起来简单但它决定了 AI 输出的边界。很多 AI 生成错误 SQL 的情况其实不是模型能力不够而是 Prompt 没有给足约束。3.2 校验层白名单与 SQL 语义检查AI 返回的 SQL在真正执行之前必须经过校验层。这一层要做到两件事先拦危险语句再做语义检查。第一步做语法级检查。把 SQL 变成小写后判断第一个关键字是不是select或with。这一步能拦截大多数 DDL、DML 语句。第二步做关键词黑名单。即使语句以 SELECT 开头也可能包含子查询、注释、联合查询等绕过手段。所以要检查是否出现insert、update、delete、drop、truncate、alter、grant、exec、copy等危险关键字。第三步去掉注释。有些恶意 SQL 会用注释符把后面的校验代码注释掉。因此在校验前要先把--注释和/* */注释剥离。第四步检查分号。一条查询语句末尾可以有分号但不能在中间出现多个分号。更稳妥的做法是校验后直接去掉末尾分号避免一次提交多条语句。第五步追加 LIMIT。在 SQL 末尾自动加上LIMIT 条数。如果 AI 生成的 SQL 已经包含 LIMIT则尽量保留原样同时检查 LIMIT 数值是否超过上限。这个步骤能有效避免一次性拉回全表数据。3.3 执行层只读连接、超时与行数限制即使 SQL 通过了校验执行时也要防一手。只读连接。数据库连接使用前面创建的低权限只读账号。这里要注意不要让连接串使用业务系统的高权限账号。超时控制。SQL 执行要有超时时间。MySQL 可以在会话级别设置max_execution_timeSQLAlchemy 连接层面也可以设置超时。对于慢查询宁可让它超时失败也不要让它占用数据库资源。结果行数限制。在代码里使用fetchmany(n)而不是fetchall()。即使 SQL 本身没有 LIMIT代码层也不能把所有数据一次性读进内存。这三层设计做完AI 查询网关的骨架已经出来了。下面进入完整实战。4. 完整实战搭建一个可运行的 AI 查询网关4.1 创建项目与依赖新建项目目录并在其中创建虚拟环境mkdir ai_query_gateway cd ai_query_gateway python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate安装依赖pip install fastapi uvicorn sqlalchemy pymysql openai4.2 准备测试数据本文以 MySQL 为例假设有一个业务库shop包含客户表和订单表。在测试环境执行下面的建表语句USE shop; CREATE TABLE customers ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL, city VARCHAR(50) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE orders ( id INT PRIMARY KEY AUTO_INCREMENT, customer_id INT NOT NULL, total_amount DECIMAL(10,2) NOT NULL, status VARCHAR(20) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_customer (customer_id), KEY idx_created (created_at) );插入几条样例数据INSERT INTO customers (name, city) VALUES (张三, 上海), (李四, 北京), (王五, 上海); INSERT INTO orders (customer_id, total_amount, status, created_at) VALUES (1, 100.00, 已完成, 2024-03-01 10:00:00), (2, 250.50, 已完成, 2024-03-05 14:30:00), (3, 80.00, 待支付, 2024-04-02 09:00:00), (1, 320.00, 已完成, 2024-04-10 20:00:00);这些数据只是用来验证程序可以跑通你可以换成自己的业务表。4.3 实现数据库连接管理文件路径database.pyimport os from sqlalchemy import create_engine # 请根据实际环境修改连接串建议使用只读账号 DATABASE_URL os.getenv( DATABASE_URL, mysqlpymysql://ai_reader:你的密码127.0.0.1:3306/shop, ) engine create_engine( DATABASE_URL, pool_pre_pingTrue, pool_recycle3600, )这里需要注意两点连接串中的账号是ai_reader这是一个只读账号不是业务主账号。pool_pre_ping会在每次取连接前先探测连接是否可用避免拿到失效连接。pool_recycle设置连接回收时间防止 MySQL 的wait_timeout断掉长连接。4.4 实现 SQL 安全校验器文件路径sql_validator.py这是整个网关中最重要的一个文件。它的核心职责是“不信任 AI 输出的任何内容”。import re SELECT_KEYWORDS {select, with} FORBIDDEN_KEYWORDS [ insert, update, delete, replace, drop, truncate, alter, create, grant, revoke, exec, execute, call, copy, merge, vacuum, ] DEFAULT_LIMIT 100 MAX_LIMIT 200 def normalize_sql(sql: str) - str: 去掉注释、字符串内容和末尾分号返回统一格式的 SQL。 sql re.sub(r--[^\n]*, , sql) sql re.sub(r/\*.*?\*/, , sql, flagsre.S) sql re.sub(r[^]*, , sql) sql sql.strip().strip(;).strip() return sql def extract_limit(sql: str) - int | None: 从 SQL 中提取 LIMIT 数值没有则返回 None。 match re.search(r\blimit\s(\d), sql, flagsre.IGNORECASE) if match: return int(match.group(1)) return None def enforce_limit(sql: str, max_rows: int MAX_LIMIT) - str: 如果 SQL 没有 LIMIT就自动追加 LIMIT。 sql sql.strip().rstrip(;).strip() if extract_limit(sql) is None: return f{sql} LIMIT {max_rows} return sql def validate_select_only(sql: str) - bool: 校验 SQL 是否为只读查询返回 True 表示通过。 normalized normalize_sql(sql) if not normalized: return False first_keyword normalized.split()[0].lower() if first_keyword not in SELECT_KEYWORDS: return False lower_sql normalized.lower() for keyword in FORBIDDEN_KEYWORDS: if re.search(rf\b{keyword}\b, lower_sql): return False # 检查是否一次提交了多条语句 if ; in normalized: return False limit extract_limit(normalized) if limit is not None and limit MAX_LIMIT: return False return True这里的关键点在于先剥离注释防止 AI 在 SQL 里加注释干扰判断。把字符串字面量替换为空避免字符串内容里夹带危险关键词。只允许select或with开头从语法层面拦截危险操作。对分号做严格检查避免堆叠多条语句。4.5 实现大模型调用层文件路径ai_client.py大模型接口每家服务商都不一样本文以 OpenAI 兼容接口为例这是目前很多大模型服务都支持的调用方式。如果你的模型服务不兼容这个格式只需要把call_llm_to_generate_sql内部实现替换成你自己的 SDK 即可。import os import re from openai import OpenAI client OpenAI( api_keyos.getenv(LLM_API_KEY, 你的 API Key), base_urlos.getenv(LLM_BASE_URL, 你的模型服务地址), ) SYSTEM_PROMPT 你是数据库查询助手。你只能根据提供的表结构生成 SELECT 查询。 要求 1. 只能使用 SELECT 或 WITH 开头。 2. 只能使用提供的表和字段。 3. 不允许出现 INSERT、UPDATE、DELETE、DROP、ALTER、TRUNCATE、GRANT 等关键字。 4. 如果问题无法用现有表结构回答请直接说明。 5. 默认限制返回不超过 200 行。 6. 只输出 SQL不要输出任何解释。 TABLE_SCHEMA 表 customers - id INT 主键 - name VARCHAR 客户姓名 - city VARCHAR 城市 - created_at DATETIME 创建时间 表 orders - id INT 主键 - customer_id INT 关联 customers.id - total_amount DECIMAL 订单金额 - status VARCHAR 订单状态 - created_at DATETIME 创建时间 def call_llm_to_generate_sql(question: str) - str: response client.chat.completions.create( modelos.getenv(LLM_MODEL, 你的模型名称), temperature0.0, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: f表结构如下\n{TABLE_SCHEMA}\n\n用户问题{question}}, ], ) sql response.choices[0].message.content.strip() # 如果模型把 SQL 放在代码块中去掉多余标记 sql re.sub(r^(sql)?, , sql).strip() sql re.sub(r$, , sql).strip() return sql这段代码需要根据你的大模型服务实际情况调整。没有标准答案关键是把握两个原则系统 Prompt 一定要写清楚“只允许 SELECT”。模型输出后要做基础清洗去掉代码块标记。4.6 实现 FastAPI 查询接口文件路径main.pyimport time from fastapi import FastAPI, HTTPException from pydantic import BaseModel from sqlalchemy import text from ai_client import call_llm_to_generate_sql from database import engine from sql_validator import ( enforce_limit, validate_select_only, DEFAULT_LIMIT, ) app FastAPI(titleAI 数据查询网关) class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): sql: str columns: list rows: list row_count: int cost_ms: int app.post(/query, response_modelQueryResponse) def query_database(req: QueryRequest): # 1. 调用大模型生成 SQL sql call_llm_to_generate_sql(req.question) # 2. 安全校验失败直接拒绝 if not validate_select_only(sql): raise HTTPException(status_code400, detailAI 生成的 SQL 未通过安全校验已拒绝执行) # 3. 自动补 LIMIT sql enforce_limit(sql, max_rows200) # 4. 使用只读连接执行查询 start time.time() try: with engine.connect() as conn: result conn.execute(text(sql)) columns list(result.keys()) rows [list(row) for row in result.fetchmany(DEFAULT_LIMIT)] except Exception as exc: raise HTTPException(status_code500, detailf数据库执行失败{str(exc)}) cost_ms int((time.time() - start) * 1000) return QueryResponse( sqlsql, columnscolumns, rowsrows, row_countlen(rows), cost_mscost_ms, )这段代码的逻辑很清晰先让 AI 生成 SQL。校验 SQL 是否安全。自动补充 LIMIT。用只读连接执行最多取 100 行。返回 SQL、表头、数据行和执行耗时。4.7 启动服务并验证在项目根目录启动 FastAPI 服务uvicorn main:app --reload --port 8000新开一个终端用 curl 测试curl -X POST http://127.0.0.1:8000/query \ -H Content-Type: application/json \ -d {question: 上海客户一共下了多少订单}预期会返回类似下面的结构{ sql: SELECT COUNT(o.id) AS order_count FROM orders o JOIN customers c ON o.customer_id c.id WHERE c.city 上海;, columns: [order_count], rows: [[2]], row_count: 1, cost_ms: 35 }你也可以在浏览器打开http://127.0.0.1:8000/docs通过 Swagger UI 直接调试接口。为了验证安全校验是否生效可以故意构造一个危险的提问比如“把 orders 表删掉”。此时 AI 生成的 SQL 应该是 DELETE 或 DROP 语句校验器会返回 400 错误不会真正执行。5. 常见问题与排查思路5.1 AI 生成的 SQL 不对怎么办这是遇到最多的一个问题。AI 生成的 SQL 本身合法但查出来的结果和预期不一致。通常原因有几种表结构信息不完整字段含义描述不到位。用户问题存在歧义比如“最近一个月”没有给出明确基准日期。模型没有理解字段之间的关联关系。解决思路把表结构描述写得更加详细尤其是字段注释和枚举值。对于常见问题可以在 Prompt 里给出几个示例问答。如果业务复杂建议在网关前增加一层“语义层”提前把术语映射成具体 SQL 片段而不是让 AI 每次重新理解。5.2 查询太慢拖垮数据库怎么办即使加了 LIMIT一个没有 WHERE 条件的大表查询依然可能很慢。解决思路在数据库账号层面设置最大执行时间。MySQL 可以在连接建立后执行SET max_execution_time 5000。在 SQLAlchemy 连接上设置超时参数。在 SQL 校验器里检查 WHERE 条件如果一条查询没有 WHERE直接拒绝或要求用户确认。为常用维度的字段建索引。索引能显著提升查询速度。把 AI 查询路由到只读从库避免影响主库业务。5.3 用户问的问题超出表结构范围怎么办比如用户问“哪个客户信用分最高”但你的表里根本没有信用分字段。这种情况下AI 可能硬编造一个字段名导致 SQL 执行报错。更合理的处理方式在 Prompt 里明确要求 AI 在字段不存在时直接说明不要强行生成 SQL。网关收到这样的响应后返回一个友好的提示而不是把报错堆给用户。5.4 权限和审计怎么做权限方面的第一原则是AI 服务使用的数据库账号权限必须小于等于业务所需的最小权限。只读账号、最小权限、网络隔离缺一不可。审计方面建议在网关里记录以下信息用户问题原文。AI 生成的 SQL。校验结果。执行耗时。返回行数。用户身份和 IP。这些日志不需要太复杂保留到日志系统即可。将来出问题可以快速回溯是哪条 SQL、哪个用户、什么时间执行的。5.5 常见问题速查问题现象常见原因解决思路AI 生成 SQL 中包含不存在字段表结构描述不完整补充字段注释和枚举值必要时增加语义层危险 SQL 未被拦截校验规则不够严格完善正则和关键字黑名单或改用 AST 解析返回数据量过大没有 LIMIT 限制自动追加 LIMIT并使用 fetchmany查询执行很慢大表全表扫描建立索引、设置超时、限制 WHERE 必须存在用户绕过网关直接连库数据库账号权限过大使用只读账号限制来源 IP模型输出带解释文字Prompt 约束不足在 Prompt 中强调只输出 SQL并对输出做清洗6. 工程化最佳实践6.1 最小权限原则在数据库层面AI 查询服务应该使用独立账号密码不能写死在代码里而是通过环境变量或配置中心注入。账号权限只保留 SELECT甚至可以只授权特定的几张表或者视图。这里要特别提一个读者经常问的问题视图可以加快查询速度吗视图本质是一段保存好的查询逻辑它不会自动提升性能但它可以隐藏底层表结构统一业务口径。在 AI 查询场景中给 AI 暴露设计良好的视图比直接暴露原始表更安全也更不容易生成错误 SQL。所以不要指望视图能优化性能但可以把它当成一组“可控的查询入口”。6.2 尽可能使用 AST 解析而不是纯文本校验纯正则做 SQL 校验有很多边界问题比如 WHERE 条件里的字符串可能包含危险关键词。更稳妥的方案是使用 SQL 解析器例如sqlglot或sqlparse把 SQL 解析成语法树后判断节点类型。示例思路如下需要按实际版本调整import sqlglot from sqlglot import exp from sqlglot.errors import ParseError def validate_sql_by_ast(sql: str) - bool: try: expressions sqlglot.parse(sql) except ParseError: return False for expression in expressions: # SQLGlot 会把非 SELECT 表达式的类型暴露出来 if not isinstance(expression, exp.Select): return False # 继续检查是否包含 DDL / DML 节点 for node in expression.walk(): if isinstance(node, (exp.Insert, exp.Update, exp.Delete, exp.Drop)): return False return True用 AST 的好处是能真正理解 SQL 结构而不是靠字符串匹配猜。大型项目里强烈建议把校验层从正则升级成解析器。6.3 审计与监控网关是天然的统一入口。每次查询都应该记录日志建议包含以下字段timestamp: 2024-05-10 10:00:00 user_id: u_1001 question: 上海客户一共下了多少订单 generated_sql: SELECT ... validated: true execution_ms: 35 row_count: 1如果条件允许可以把这些日志同步到日志中心或数据库用于后续分析。监控方面重点关注失败率、平均耗时、被拦截的 SQL 数量。被拦截的次数突然上升往往说明 Prompt 需要调整或者用户开始尝试输入危险指令。6.4 Prompt 工程设计要持续迭代很多人把 Prompt 当成一次写完就固定下来的配置这是不对的。Prompt 需要根据真实问题持续迭代。一个建议的做法是先在测试库上跑一批典型问题把 AI 生成的错误 SQL 收集起来分析是表结构描述不清楚还是约束条件不明确再针对性地修改 Prompt。对于特别容易出错的问题可以在 Prompt 末尾追加“常见问题示例”。这个迭代过程比频繁更换大模型更有效。6.5 向量检索和语义缓存的扩展思路当你的业务表结构很多、常见的查询问题重复出现时可以考虑引入向量数据库。把“自然语言问题”和“正确 SQL”成对存入向量库收到新问题时先用向量检索召回相似的 SQL再让大模型基于召回结果做修改。这样做有两个好处一是降低大模型生成错误 SQL 的概率二是减少大模型调用成本。对高频问题可以完全走缓存不需要每次都调用模型。7. 总结与下一步学习方向这套 AI 查询网关的完整链路核心只有三句话AI 只负责生成校验必须独立执行必须受限。数据库权限用只读账号SQL 校验做白名单和关键词拦截执行时加超时和行数限制再把每一步都记录下来。有了这层网关你才真正可以放心地把数据库查询交给 AI。下一步建议你重点深入研究几个方向SQL 语法树解析把校验层从正则升级到 AST这是生产级方案的必经之路。语义层的构建用视图或中间表统一业务口径让 AI 面对的不再是一堆裸表。模型选型与成本控制选择适合 SQL 生成的模型评估准确率、延迟和调用成本。生产环境的隔离策略如果条件允许把 AI 查询指向只读从库与主库完全隔离。先把测试库跑通再逐步放开权限和场景。AI 查询这条路并不神秘但要靠一层一层的约束才能让它在真实业务中安全落地。如果这篇文章对你有帮助可以收藏备用后续有更好的工程实践我会继续补充。