企业官网建设流程全解析

这次我们来看一个矛盾点：AI 大模型写 SQL 很快，但谁敢让它在生产库上跑？

Text-to-SQL 的价值不需要多解释。开发者问一句“上个月订单量前 10 的商品是什么”，模型就能生成一条带 GROUP BY、ORDER BY、HAVING 的 SQL，甚至直接返回结果。问题从来不是“能不能写”，而是“写出来之后你怎么确认它不会把全表 UPDATE 成 NULL，不会把几千万行的表全扫一遍，不会因为取了错误字段直接报错”。

这次我们说的这个方案，核心思路就是把数据库查询交给 AI 之前，先确定几件事：权限是否收敛、查询是否只读、SQL 是否经过校验、执行有没有审计、超时和限流有没有兜底。它不是一个某某公司开源的固定仓库，而是一套“AI + 数据库查询”的工程化落地思路，结合了当前主流的 Text-to-SQL 工具链、RAG 表结构检索、LangChain 和 Spring AI 等调用方式。

如果你正在做 AI Agent、数据库问答、内部数据分析平台，或者只是在调研“能不能把自然语言查询接到现有系统上”，这篇文章值得收藏。全文会按这样的路径走：先用一张表把核心能力说清楚，然后给出前置条件和环境准备，再拆解安全设计、部署启动、功能测试、API 接入、批量任务、资源占用和排错清单，最后是合规和最佳实践。重点不是概念，而是你能照着跑通并判断这套东西适不适合你的场景。

1. 核心能力速览

从这套能力看，它解决的核心不是“生成 SQL”，而是“生成 SQL 之后怎么安全地执行”。很多项目都死在最后一步：模型生成了错误的表名、错误的字段名，或者生成了 UPDATE 语句，直接改坏数据。所以，下面所有章节都会围绕“安全查询”展开。

2. 适用场景与使用边界

2.1 适合谁

• 数据分析师：用自然语言问业务问题，减少手写复杂 SQL 的时间。
• 后端开发者：把自然语言查询封装成接口，提供给前端报表或内部工具。
• AI Agent 开发者：让 Agent 具备查询业务数据库的能力，而不用把数据库凭据暴露给模型层。
• 企业内部知识库建设：把表结构、字段注释、枚举值说明喂给模型，提高 SQL 生成准确率。

2.2 能解决什么问题

• 降低 SQL 编写门槛，让业务人员直接问数。
• 统一查询入口，避免每个部门各自连库。
• 支持批量问题查询，适合周报、月报、数据运营场景。
• 通过权限收敛只读账号，减少误操作风险。

2.3 不适合什么场景

• 高危写操作：比如批量 UPDATE、DELETE，不应该让 AI 直接执行。
• 超大规模查询：如果一张表有几十亿行，模型生成的 SQL 极容易全表扫描，需要额外加 LIMIT 或查询超时。
• 敏感数据直接开放：手机号、身份证、财务明细等字段，需要做脱敏和权限分级，不能全部暴露给 AI 查询链路。
• 无审计的正式环境：生产库必须接审计日志，否则出了问题无法追溯。

2.4 合规提醒

如果接入数据库查询能力，必须注意几个边界：

• 只读账号是最低底线，不要把写权限交给 AI。
• 数据库内如果有用户隐私数据、企业经营数据，需要确认查询行为符合内部数据安全规范。
• 如果使用本地大模型处理 SQL，需要关注模型文件本身的合规来源。
• 任何涉及人脸、电话、地址等敏感字段的查询，都应该有字段级权限控制和脱敏机制。

3. 环境准备与前置条件

下面给出一套通用检查清单，适合大多数 Text-to-SQL 项目。

3.1 操作系统与软件环境

3.2 大模型选择

两种方式：

• API 模式：适合快速验证，不需要本地 GPU。准备 API Key 和接口地址。
• 本地模型模式：需要本地部署推理服务，例如 vLLM、Ollama、Xinference 等，需要一块显存足够的显卡，具体显存占用按模型大小和量化方式决定，不能一概而论。

3.3 数据库账号准备

不要把 DBA 账号直接给 AI。单独创建一个只读账号：

-- MySQL 示例：创建只读账号 CREATE USER 'ai_query'@'%' IDENTIFIED BY 'your_password'; GRANT SELECT ON your_database.* TO 'ai_query'@'%'; FLUSH PRIVILEGES;

-- PostgreSQL 示例 CREATE USER ai_query WITH PASSWORD 'your_password'; GRANT CONNECT ON DATABASE your_database TO ai_query; GRANT USAGE ON SCHEMA public TO ai_query; GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_query;

这里的关键点：只给 SELECT。不要让 AI 生成的 SQL 有 UPDATE、DELETE、INSERT、DDL 的能力。这是整个方案里最便宜也最有效的安全措施。

3.4 表结构信息准备

模型要生成准确 SQL，必须知道表名、字段名、字段类型、字段注释、枚举值。建议导出一份表结构说明，写入知识库或作为 Prompt 上下文。

# MySQL 导出表结构信息 mysqldump -u ai_query -p --no-data your_database > schema.sql

也可以查询 information_schema：

SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = 'your_database' ORDER BY TABLE_NAME, ORDINAL_POSITION;

4. 系统架构与安全设计

如果是自己搭这套“AI 查询数据库”的服务，建议按下面的分层设计。

4.1 架构分层

用户输入 ↓ 自然语言处理层（大模型 + Prompt 模板 + 表结构上下文） ↓ SQL 生成层（Text-to-SQL 模型或大模型函数调用） ↓ SQL 安全校验层（只读检查、关键字拦截、语法解析、表名白名单） ↓ 执行层（只读账号、超时控制、LIMIT 强制、审计日志） ↓ 结果返回层（格式化、脱敏、缓存）

每一层都不能少。尤其是 SQL 安全校验层，这是整个项目能不能落地的关键。

4.2 安全校验规则

至少要校验以下几项：

1. 是否只包含 SELECT。
2. 不包含 INTO OUTFILE、LOAD_FILE、SLEEP、BENCHMARK 等危险函数。
3. 不包含 INFORMATION_SCHEMA 之外的系统库访问。
4. 强制追加 LIMIT，默认 100 条。
5. 表名必须来自白名单。
6. 字段名校验，避免模型生成不存在的字段。

示例校验逻辑：

import re FORBIDDEN_KEYWORDS = [ "update", "delete", "insert", "drop", "alter", "truncate", "grant", "revoke", "create", "replace", "load_file", "into outfile", "sleep", "benchmark", "information_schema" ] def validate_sql(sql: str) -> bool: sql_lower = sql.lower().strip() if not sql_lower.startswith("select"): return False for kw in FORBIDDEN_KEYWORDS: if re.search(rf"\b{kw}\b", sql_lower): return False return True

5. 安装部署与启动方式

这里给两套参考实现：一套基于 Python + LangChain，一套基于 Java + Spring AI。你可以根据团队技术栈选择。

5.1 Python + LangChain 方式

创建虚拟环境并安装依赖：

python -m venv venv source venv/bin/activate pip install langchain langchain-community langchain-openai pip install pymysql sqlalchemy

准备数据库连接字符串和模型配置。下面是一个最小示例：

from langchain.agents import create_sql_agent from langchain.agents.agent_toolkits import SQLDatabaseToolkit from langchain.sql_database import SQLDatabase from langchain_openai import ChatOpenAI # 数据库连接，注意使用只读账号 db = SQLDatabase.from_uri( "mysql+pymysql://ai_query:your_password@127.0.0.1:3306/your_database", include_tables=["orders", "products", "users"], sample_rows_in_table_info=3 ) # 大模型，可替换为本地模型地址 llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, base_url="https://api.openai.com/v1", api_key="your_api_key" ) toolkit = SQLDatabaseToolkit(db=db, llm=llm) agent = create_sql_agent( llm=llm, toolkit=toolkit, verbose=True, handle_parsing_errors=True ) # 执行查询测试 result = agent.invoke("查询最近7天每个商品的订单数量，按订单数量降序排列，只要前10条") print(result)

这里注意几个参数：

• include_tables只让模型看到指定的表，减少幻觉。
• sample_rows_in_table_info让模型知道真实数据的示例格式。
• temperature=0避免模型自由发挥，SQL 生成必须尽量确定。

5.2 Java + Spring AI 方式

如果团队是 Java 技术栈，Spring AI 提供了类似的能力。先引入依赖：

<dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-starter-model-openai</artifactId> <version>插入当前版本</version> </dependency> <dependency> <groupId>org.springframework.ai</groupId> <artifactId>spring-ai-starter-vector-store</artifactId> <version>插入当前版本</version> </dependency>

定义一个查询服务，核心逻辑是把表结构信息拼到 Prompt 里，再让大模型返回 SQL。这里不展开完整代码，但思路是：

1. 启动时从数据库读取字段元数据。
2. 构建系统 Prompt，包含表结构、字段注释、示例值。
3. 用户输入问题后，调用大模型接口生成 SQL。
4. 在 Java 层执行安全校验。
5. 通过 JdbcTemplate 执行 SQL，并强制设置查询超时。

5.3 Docker 部署

如果你已经有打包好的服务，推荐用 Docker Compose 维护。一个标准配置长这样：

version: "3.9" services: ai-query-service: build: . ports: - "8080:8080" environment: SPRING_AI_OPENAI_API_KEY: ${OPENAI_API_KEY} SPRING_AI_OPENAI_BASE_URL: ${OPENAI_BASE_URL} DB_URL: jdbc:mysql://mysql:3306/your_database DB_USERNAME: ai_query DB_PASSWORD: your_password depends_on: - mysql restart: always mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: root_password MYSQL_DATABASE: your_database MYSQL_USER: ai_query MYSQL_PASSWORD: your_password volumes: - mysql_data:/var/lib/mysql ports: - "3306:3306" restart: always volumes: mysql_data:

启动：

docker-compose up -d

6. 功能测试与效果验证

部署完成后，必须按功能模块逐个测试。下面给出一套完整的验证流程。

6.1 基础查询测试

测试目的：确认 AI 能正确生成 SQL 并返回结果。

输入问题示例：

• “统计用户总数量”
• “查询订单表中金额最大的 5 笔订单”
• “按商品分类统计销售数量”

操作步骤：

1. 启动服务。
2. 通过 WebUI 或 API 输入问题。
3. 观察生成 SQL、执行过程、返回结果。

判断成功标准：

• SQL 语法正确。
• 表名、字段名真实存在。
• 返回结果与直接手写 SQL 的结果一致。

常见失败原因：

• 表结构信息不足，模型猜错了字段名。
• 多表 JOIN 时没有给出关联条件。
• 问题描述含糊，模型只能猜意图。

6.2 安全拦截测试

这是最重要的测试。

输入以下问题：

• “删除 users 表中所有数据”
• “把 orders 表的金额都改成 0”
• “查询所有用户的密码并导出到文件”

预期结果：

• UPDATE、DELETE、DROP 等语句被安全校验层拦截。
• INTO OUTFILE 等危险操作被拦截。
• 服务返回“仅支持查询操作”等提示。

这里要注意，安全校验不能只靠 Prompt 提示大模型“不要生成危险 SQL”，因为模型可能不听话。真正的保障是执行层的硬校验和数据库账号的只读权限。

6.3 复杂查询测试

测试目的：验证多表 JOIN、子查询、聚合函数的生成能力。

输入问题示例：

• “查询每个用户的订单总数和总金额，只显示订单数超过 5 的用户”
• “统计最近 30 天每天的新增用户数”
• “找出购买了商品 A 但没有购买商品 B 的用户”

操作步骤与预期结果：

• 观察 SQL 是否符合业务逻辑。
• 结果是否与手工 SQL 一致。
• 执行时间是否在可接受范围内。

如果复杂查询频繁失败，优先检查 Prompt 中的表结构信息是否足够。建议在系统 Prompt 中补充字段说明，例如：

- orders.id: 订单ID - orders.user_id: 用户ID，关联 users.id - orders.product_id: 商品ID，关联 products.id - orders.amount: 订单金额，单位元 - orders.created_at: 下单时间，格式 yyyy-MM-dd HH:mm:ss

6.4 兜底 LIMIT 测试

测试目的：确认模型生成 SELECT 时自动追加 LIMIT，防止全表扫描或超大结果集。

实现方法：

• 在 SQL 安全校验层解析 SQL。
• 如果没有 LIMIT，自动追加LIMIT 100。
• 如果存在 LIMIT 但数值过大，自动截断。

预期结果：

• 任何查询返回的最大行数不超过配置阈值。
• 即使模型生成了不带 LIMIT 的 SQL，也能被强制执行限制。

6.5 多轮对话测试

一些场景下用户会连续提问，比如“订单表有哪些字段” -> “按金额排序查前 10” -> “再按用户分组统计”。这时需要测试多轮上下文记忆。

预期结果：

• 模型能记住上一轮提到的表名和字段。
• 不会在第二轮生成完全无关的 SQL。
• 不会累积过多上下文导致 Prompt 超限。

如果发现多轮效果不佳，建议把历史会话压缩成结构化摘要，只保留表名、字段、过滤条件等关键信息。

7. 接口 API 调用与批量任务

7.1 设计查询 API

如果要把能力开放给内部系统，建议提供一个 POST 接口。

接口请求示例：

{ "query": "查询最近7天每个商品的订单数量，按订单数量降序排列，只要前10条", "session_id": "session_001", "max_rows": 50 }

响应示例：

{ "success": true, "sql": "SELECT p.product_name, COUNT(o.id) AS order_count FROM orders o JOIN products p ON o.product_id = p.id WHERE o.created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY) GROUP BY p.product_name ORDER BY order_count DESC LIMIT 10", "columns": ["product_name", "order_count"], "rows": [ ["商品A", 128], ["商品B", 96] ], "execution_time_ms": 156 }

这种结构适合前端直接渲染表格。

7.2 curl 调用示例

curl -X POST http://127.0.0.1:8080/api/query \ -H "Content-Type: application/json" \ -H "Authorization: Bearer your_token" \ -d '{ "query": "查询库存低于10的商品列表", "session_id": "test_001", "max_rows": 20 }'

7.3 Python 调用示例

import requests url = "http://127.0.0.1:8080/api/query" payload = { "query": "查询最近30天每天的新增用户数", "session_id": "batch_001", "max_rows": 100 } headers = { "Authorization": "Bearer your_token", "Content-Type": "application/json" } response = requests.post(url, json=payload, headers=headers, timeout=60) data = response.json() if data.get("success"): print(data["columns"]) for row in data["rows"]: print(row) else: print("查询失败:", data.get("error"))

7.4 批量查询任务

批量查询适合“一批问题清单，一次跑完”的场景。建议设计一个任务队列：

[ {"query": "查询本月每日销售额", "max_rows": 50}, {"query": "查询退货率最高的10个商品", "max_rows": 50}, {"query": "查询最近7天新增用户的地域分布", "max_rows": 50} ]

处理逻辑：

1. 逐条调用查询服务。
2. 每条任务记录入参、SQL、错误信息。
3. 单条失败不影响后续任务。
4. 全部执行完后输出结果汇总。

示例 Python 批量处理脚本：

import json import time import requests API_URL = "http://127.0.0.1:8080/api/query" TOKEN = "your_token" queries = json.load(open("batch_queries.json", "r", encoding="utf-8")) results = [] for idx, item in enumerate(queries, start=1): try: resp = requests.post( API_URL, json=item, headers={"Authorization": f"Bearer {TOKEN}"}, timeout=90 ) resp_data = resp.json() results.append({ "index": idx, "query": item["query"], "success": resp_data.get("success", False), "error": resp_data.get("error", ""), "sql": resp_data.get("sql", ""), "rows_count": len(resp_data.get("rows", [])) if resp_data.get("rows") else 0 }) print(f"[{idx}] {'OK' if results[-1]['success'] else 'FAIL'} - {item['query']}") except Exception as e: print(f"[{idx}] ERROR - {item['query']} - {e}") time.sleep(0.5) with open("batch_results.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) print("全部任务执行完毕，结果已写入 batch_results.json")

批量任务建议加上以下机制：

• 每次任务之间加延迟，避免接口被打满。
• 对单条失败的任务做 2 次重试。
• 日志里记录每条任务执行的 SQL，方便审计。

8. 性能与资源占用观察

8.1 大模型对性能的影响

这个项目的大头性能消耗取决于大模型怎么接：

• 纯 API 模式：本地只需要运行查询服务本身，资源占用主要看并发量和数据库查询量。
• 本地大模型模式：需要单独部署推理服务，显存占用取决于模型参数量和量化级别。一个 7B 模型用 4bit 量化通常需要 6G 到 8G 显存，13B 模型可能需要 10G 到 16G 显存。具体数字要按实际模型和推理框架测试。

8.2 数据库查询的性能瓶颈

即使 AI 生成 SQL 再快，真正执行时还是会回到数据库本身。重点观察：

• 慢查询日志里有没有出现新的全表扫描。
• JOIN 的表是否有索引。
• 查询结果集是否超过预期。
• 最大行数限制是否生效。

8.3 如何观察资源占用

# 观察 GPU 显存占用 nvidia-smi -l 2 # 观察服务进程 CPU 和内存 top -p $(pgrep -f ai_query_service) # 数据库慢查询日志 tail -f /var/log/mysql/mysql-slow.log

8.4 常见性能问题与处理

9. 常见问题与排查方法

如果启动后服务端口打不开，先检查端口占用：

# 查看端口占用情况 lsof -i :8080 # 或者 netstat -tunlp | grep 8080

如果端口冲突，换一个端口启动，或者修改配置文件中的端口号。

如果模型文件缺失，本地模型模式下会看到类似 “file not found” 的错误。确认模型文件路径和名称是否与实际下载的一致，注意可能还需要一个额外的配置文件描述模型路径和量化格式。

如果 Python 依赖安装失败，优先检查 Python 版本和 pip 源。常见解决办法：

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

如果 Spring AI 配置不生效，确认环境变量和 application.yml 是否匹配。重点检查：

• API Key 有没有写对。
• Base URL 是否包含/v1路径。
• 数据库连接串的时区参数是否需要补充。

10. 最佳实践与使用建议

10.1 第一次使用先跑最小集

不要一上来就接 200 张表。先选 5 到 10 张核心表，让模型生成 SQL，人工核对正确率。正确率稳定在 90% 以上再扩展表范围。

10.2 把表结构信息做成可维护的元数据

不要每次启动都重新查数据库字段。建议形成一份元数据文件，包含表名、字段、类型、注释、枚举值、表间关联关系。这份文件就是给模型看的“数据库说明书”。

[ { "table_name": "orders", "table_comment": "订单表", "columns": [ {"name": "id", "type": "bigint", "comment": "订单ID"}, {"name": "user_id", "type": "bigint", "comment": "用户ID，关联 users.id"}, {"name": "product_id", "type": "bigint", "comment": "商品ID，关联 products.id"}, {"name": "amount", "type": "decimal(10,2)", "comment": "订单金额，单位元"}, {"name": "created_at", "type": "datetime", "comment": "下单时间"} ], "sample_values": {} } ]

这份文件可以用脚本从数据库自动生成，也可以手工维护。表结构变更后记得同步更新。

10.3 每个查询都要留审计证据

正式环境应该记录：谁问了什么问题、模型生成了什么 SQL、SQL 实际执行了多久、返回了多少行。这样一旦出问题，可以回溯到用户和 SQL，不会变成“死无对证”。

10.4 接口服务要限制访问范围

不要把查询 API 暴露到公网。建议：

• 只在内网调用。
• 加 API Token 或跳板认证。
• 按调用方做限流。

10.5 涉及敏感字段必须脱敏

查询结果返回前，对手机号、身份证、邮箱等字段做脱敏处理。常见做法：

def mask_phone(phone: str) -> str: if not phone or len(phone) < 7: return phone return phone[:3] + "****" + phone[-4:]

10.6 发布前做一轮完整效果复核

正式开放给业务方之前，准备 50 到 100 条典型问题，跑一遍，统计：

• 查询正确率。
• 平均响应时间。
• 失败率。
• 危险 SQL 拦截率。

把这些指标整理成文档，团队内部确认后再上线。

收尾

把数据库查询交给 AI，真正值得信任的方式不是让大模型“随便生成然后碰运气”，而是形成一条“生成 SQL -> 安全校验 -> 只读执行 -> 审计记录”的链路。权限收敛是底线，安全校验是防线，表结构元数据质量决定了准确率上限。对这个项目最值得先做的三件事：第一，把数据库账号切到只读；第二，写一个 SQL 安全校验函数；第三，整理出前 10 张核心表的元数据。这三步做完，整个系统的风险已经降了一大半。后续再逐步加缓存、接口限流、批量任务和结果导出，就是一个可以放进内网正式使用的生产力工具。