任务清单

1. 项目环境搭建

部署 openEuler 服务器，关闭 SELinux、配置可用 yum 源，编译安装 Python3.11.9；

安装 MySQL8.0.45，创建 testdb 库、order_info 订单业务表，导入 10 条标准测试订单数据；

登录并开通腾讯云 TokenHub 大模型聚合平台，申请 API 密钥，领取免费额度，编写.env 配置文件并加固

文件权限；

安装项目全部 Python 依赖包，测试数据库连通性、大模型接口连通性。

2. 脚本开发

mysql_client.py：封装 MySQL 连接、查询与执行计划获取，拦截危险 SQL 保障数据库安全。

prompts.py：统一存放生成 SQL、调优分析两套 AI 提示词，并提供提取纯净 SQL 的工具函数。

main.py：项目总调度，串联数据库、大模型、提示词，提供终端菜单交互，实现完整业务逻辑。

web_main.py：Streamlit 可视化网页界面，做输入校验、图形化展示结果，复用 main 核心逻辑。

.env：存放数据库、腾讯云 TokenHub 大模型密钥等配置，分离敏感信息与业务代码。

3. 功能测试

多业务自然语言用例全覆盖测试，校验生成 SQL 安全、逻辑准确，模糊匹配无 bug；

性能调优模块测试，校验 EXPLAIN 解析完整、输出索引与改写方案可直接执行；

边界场景测试：无匹配用户 / 商品、空白输入、超长复合需求，程序无崩溃报错；

安全测试：验证不会生成任何 DELETE/UPDATE/ALTER 等危险 SQL，密钥无明文打印。

实验过程

依赖安装

sed -i '7s/enforcing/disabled/' /etc/selinux/config

systemctl disable --now firewalld

systemctl status firewalld

dnf install -y gcc gcc-c++ make cmake zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel libffi-devel tk-devel wget tar vim tree net-tools openssh-server

cd /usr/local/src

tar -zxvf Python-3.11.9.tgz

cd Python-3.11.9

./configure --prefix=/usr/local/python3.11 --enable-shared

make -j$(nproc) && make install

echo "/usr/local/python3.11/lib" > /etc/ld.so.conf.d/python311.conf

ldconfig

ln -s /usr/local/python3.11/bin/python3.11 /usr/local/bin/python3

ln -s /usr/local/python3.11/bin/pip3.11 /usr/local/bin/pip3

python3 -V

pip3 -V

python3 -c "import ssl; print(ssl.OPENSSL_VERSION)"

cd

mkdir .pip

vim ~/.pip/pip.conf

[global]
index-url = http://mirrors.aliyun.com/pypi/simple/
[install]
trusted-host=mirrors.aliyun.com

:wq

pip3 install --upgrade pip

pip3 install pymysql python-dotenv tabulate langchain langchain-openai

mysql -u root -p

MYSQL命令

create database testdb;

use testdb;

-- 创建订单业务表
create table order_info(
id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '订单ID',
user_id INT COMMENT '用户ID',
order_name VARCHAR(200) COMMENT '商品名称',
pay_amount DECIMAL(10,2) COMMENT '支付金额',
create_time DATETIME COMMENT '下单时间'
) ENGINE=InnoDB COMMENT='电商订单业务表';

-- 导入20条标准测试数据
INSERT INTO order_info(user_id,order_name,pay_amount,create_time)
VALUES
(1001,'智能手机',2999.00,'2026-05-01 10:20:00'),
(1001,'有线入耳耳机',199.00,'2026-05-02 14:10:00'),
(1002,'14英寸轻薄笔记本电脑',5499.00,'2026-05-03 09:30:00'),
(1002,'无线蓝牙鼠标',89.00,'2026-05-03 09:35:00'),
(1003,'平板学习机',1799.00,'2026-05-04 11:05:00'),
(1003,'平板专用保护壳',49.00,'2026-05-04 11:08:00'),
(1004,'机械游戏键盘',349.00,'2026-05-05 16:42:00'),
(1004,'电竞头戴耳机',459.00,'2026-05-05 16:48:00'),
(1005,'大屏智能电视',3299.00,'2026-05-06 08:15:00'),
(1005,'电视壁挂支架',129.00,'2026-05-06 08:20:00'),
(1006,'无线快充充电器',129.00,'2026-05-03 13:22:00'),
(1006,'降噪蓝牙耳机',399.00,'2026-05-03 13:25:00'),
(1007,'电竞显示器',1899.00,'2026-05-07 10:10:00'),
(1007,'显示器增高支架',79.00,'2026-05-07 10:15:00'),
(1008,'折叠平板支架',39.00,'2026-05-04 15:30:00'),
(1008,'便携充电宝',159.00,'2026-05-04 15:33:00'),
(1009,'台式游戏主机',6999.00,'2026-05-08 09:05:00'),
(1009,'电竞防滑鼠标垫',59.00,'2026-05-08 09:08:00'),
(1010,'手机钢化膜',29.00,'2026-05-05 17:12:00'),
(1010,'桌面收纳支架',45.00,'2026-05-05 17:16:00');

-- 验证数据
select * from order_info;

exit;

脚本开发

cd

mkdir -p /opt/mysql_ai_tools

cd /opt/mysql_ai_tools

touch main.py mysql_client.py prompts.py web_main.py .env

tree -La 1

cd

vim /opt/mysql_ai_tools/.env

MYSQL_HOST=127.0.0.1
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=123456
MYSQL_DB=testdb
LLM_API_KEY=sk-jUJebmj56mEQhh*******5iimZ***
LLM_BASE_URL=https://tokenhub.tencentmaas.com/v1
LLM_MODEL_NAME=deepseek-v4-pro
LLM_TEMPERATURE=0

:wq

chmod 600 /opt/mysql_ai_tools/.env

vim /opt/mysql_ai_tools/mysql_client.py

import pymysql
import os
import re
from dotenv import load_dotenv

load_dotenv()

class Mysql80Client:
    def __init__(self):
        self.host = os.getenv("MYSQL_HOST", "127.0.0.1")
        self.port = int(os.getenv("MYSQL_PORT", "3306"))
        self.user = os.getenv("MYSQL_USER", "root")
        self.password = os.getenv("MYSQL_PASSWORD", "")
        self.database = os.getenv("MYSQL_DB", "testdb")
        self.conn = None
        self.connect()

def connect(self):
    try:
        self.conn = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,
            password=self.password,
            database=self.database,
            charset='utf8mb4', 
            cursorclass=pymysql.cursors.DictCursor 
        )
    except pymysql.MySQLError as e:
        raise Exception(f"数据库连接失败，请检查地址/账号/密码：{e.args[1]}")
    except Exception as e:
        raise Exception(f"数据库连接异常：{str(e)}")

@staticmethod
def _check_sql_safety(sql: str) -> None:
    """
    静态私有安全校验方法
    核心防护：拦截增删改、建表删表等危险操作，仅允许SELECT查询，防止AI生成危险SQL篡改数据
    """
    sql_trim = sql.strip().upper()
    danger_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE","TRUNCATE", "REPLACE"]
    for kw in danger_keywords:
        if re.search(r'\b' + re.escape(kw) + r'\b', sql_trim):
            raise Exception(f"安全拦截：禁止执行 {kw} 类型语句，仅支持 SELECT 查询")

def execute_query(self, sql: str):
    """
    执行普通SELECT查询
    :param sql: 待执行查询语句
    :return: (字段名列表, 全部数据行字典列表)
    """
    self._check_sql_safety(sql)
    try:
        with self.conn.cursor() as cursor:
            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
            return columns, rows
    except pymysql.MySQLError as e:
        raise Exception(f"SQL执行失败（错误码 {e.args[0]}）：{e.args[1]}")
    except Exception as e:
        raise Exception(f"查询异常：{str(e)}")

def get_explain_plan(self, sql: str):
    """
    获取SQL执行计划EXPLAIN，用于性能调优分析
    :param sql: 待分析SELECT语句
    :return: (执行计划表头, 执行计划详情数据)
    """
    self._check_sql_safety(sql)
    explain_sql = f"EXPLAIN {sql}"
    try:
        with self.conn.cursor() as cursor:
            cursor.execute(explain_sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
            return columns, rows
    except pymysql.MySQLError as e:
        raise Exception(f"获取执行计划失败：{e.args[1]}")
    except Exception as e:
        raise Exception(f"执行计划异常：{str(e)}")

def close(self):
    """安全关闭数据库连接，释放资源，避免长时间占用连接池"""
    if self.conn and not self.conn._closed:
        self.conn.close()

:wq

vim /opt/mysql_ai_tools/prompts.py

import re

class UnifiedPrompt:
    """
    提示词统一管理类
    优势：所有SQL生成、性能分析提示词集中存放，表结构仅维护一处，修改不用多处同步；
    通过严格规则约束大模型输出，减少格式错乱、编造字段、危险SQL等幻觉问题
    """

    TABLE_SCHEMA = """
    表名: order_info (订单信息表)
    字段说明:
    - id: 订单ID (主键，INT类型)
    - user_id: 用户ID (INT类型)
    - order_name: 商品名称 (VARCHAR类型)
    - pay_amount: 支付金额 (DECIMAL类型)
    - create_time: 下单时间 (DATETIME类型)
    """

    NL_TO_SQL_PROMPT = f"""
    你是严谨的 MySQL 8.0 数据库开发工程师。
    【任务目标】
    根据用户自然语言描述的业务需求，生成可直接执行、无语法错误的MySQL查询SQL。
    【表结构参考】
    {TABLE_SCHEMA}
    【强制输出规则】
    1. 只能生成 SELECT 查询语句，绝对不允许生成 INSERT/UPDATE/DELETE/DROP 等修改、删除数据的语句。
    2. 只能使用上面列出的5个字段，禁止自己编造不存在的字段名。
    3. 查询字段可使用中文别名，格式固定为：字段 AS 别名。
    4 SQL语法遵循MySQL8.0标准，所有关键字统一大写，方便程序解析。
    5. 最终SQL必须包裹在 ```sql ``` Markdown代码块内，方便代码提取。
    6. 禁止输出任何解释、说明文字，只返回纯SQL代码块，减少解析干扰。
    7. 中文别名内部不能带空格，例：订单ID（正确）、订单 ID（错误），避免数据库语法报错。
    【用户需求】
    {{user_input}}
    """

    SQL_TUNE_PROMPT = f"""
    你是资深 MySQL DBA 性能优化专家。
    【任务目标】
    根据原始SQL + EXPLAIN执行计划数据，定位查询性能问题并给出可直接落地的优化方案。
    【表结构参考】
    {TABLE_SCHEMA}
    【待分析SQL】
    {{sql_input}}
    【执行计划数据】
    {{explain_data}}
    【输出要求】
    1. 先点明核心性能问题：全表扫描、无索引、索引失效、扫描行数过多等。
    2. 给出完整建索引SQL语句，可直接复制执行。
    3. 若原SQL写法存在缺陷，提供改写后的完整优化SQL。
    4. 内容简洁、分点罗列，不输出多余废话，便于用户快速阅读。
    """
    @staticmethod
    def extract_sql(response_text: str) -> str:
    """
    静态工具方法：从大模型返回的完整文本里剥离出纯净SQL语句
    三层匹配优先级，兼容不同大模型的输出格式，提升提取成功率
    :param response_text: 大模型原始完整返回内容
    :return: 清洗后的纯SQL字符串，提取失败返回空字符串
    """
    if not response_text:
    return ""

    match = re.search(r"```sql\s*(.*?)\s*```", response_text, re.DOTALL | re.IGNORECASE)
    if match:
        return match.group(1).strip()

    match = re.search(r"<sql>\s*(.*?)\s*</sql>", response_text, re.DOTALL | re.IGNORECASE)
    if match:
        return match.group(1).strip()

    match = re.search(r"(SELECT\s+.*?;)", response_text, re.DOTALL | re.IGNORECASE)
    if match:
        return match.group(1).strip()

    return ""

prompt_helper = UnifiedPrompt()

:wq

vim /opt/mysql_ai_tools/main.py

import os
import re
import logging
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mysql_client import Mysql80Client
from tabulate import tabulate
from prompts import UnifiedPrompt, prompt_helper

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - % (message)s")
logger = logging.getLogger(__name__)

def check_config() -> None:
    """
    程序启动前置配置校验函数
    作用：提前检测.env必填参数是否存在，避免运行中途缺参数崩溃
    """

    required_llm = ["LLM_API_KEY", "LLM_BASE_URL", "LLM_MODEL_NAME"]
    missing = [k for k in required_llm if not os.getenv(k)]
    if missing:
        raise ValueError(f"配置缺失：请在 .env 文件中填写 {', '.join(missing)}")

    required_db = ["MYSQL_HOST", "MYSQL_USER", "MYSQL_DB"]
    missing_db = [k for k in required_db if not os.getenv(k)]
    if missing_db:
        raise ValueError(f"数据库配置缺失：请检查 {', '.join(missing_db)}")

def get_llm() -> ChatOpenAI:
"""
初始化大模型客户端
适配腾讯云TokenHub等全部兼容OpenAI接口规范的MaaS平台
返回：可直接调用的大模型实例
"""
# 从环境变量读取大模型连接信息
api_key = os.getenv("LLM_API_KEY")
base_url = os.getenv("LLM_BASE_URL")
model_name = os.getenv("LLM_MODEL_NAME")
# 温度不存在则默认0.1，数值越低输出越严谨稳定
temperature = float(os.getenv("LLM_TEMPERATURE", 0.1))
return ChatOpenAI(
api_key=api_key,
base_url=base_url,
model=model_name,
temperature=temperature
)
def clean_sql_spacing(sql: str) -> str:
"""
SQL标准化清洗工具函数（兜底修复各大模型输出格式）
解决：中文空格别名、中文标点、特殊空白、关键字连写等语法报错问题
入参：大模型原始SQL字符串
返回：清洗后可直接执行的标准英文SQL
"""
if not sql:
return ""
# 1. 统一替换各类中文全角空格、换行、制表符为普通半角空格
special_spaces = [
'\xa0', '\u200b', '\u200c', '\u200d', '\u200e', '\u200f',
'\u3000', '\t', '\n', '\r'
]
for sp in special_spaces:
sql = sql.replace(sp, ' ')
# 2. 删除不可见控制字符，防止解析异常
sql = re.sub(r'[\x00-\x1f\x7f]', '', sql)
# 3. 中文标点批量替换为英文标点（解决Qwen等模型输出中文逗号报错）
sql = sql.replace('，', ',').replace('；', ';').replace('（', '(').replace('）', ')')
# 4. 多个连续空格合并为单个，去除首尾多余空格
sql = re.sub(r'\s+', ' ', sql).strip()
# 5. 精准处理AS别名内部空格，只删别名里空格，保留AS与别名之间分隔空格
def _clean_alias_space(match):
prefix = match.group(1) # 捕获AS关键字
alias = match.group(2) # 捕获后面全部别名文本
alias_clean = re.sub(r'\s+', '', alias)
return f"{prefix} {alias_clean}"
# 匹配AS后别名，截止逗号、FROM、WHERE等关键字前停止匹配
sql = re.sub(
r'\b(AS)\s+(.+?)(?
=\s*,\s*|\s+FROM\b|\s+WHERE\b|\s+ORDER\b|\s+GROUP\b|\s+LIMIT\b|\s*;)',
_clean_alias_space,
sql,
flags=re.IGNORECASE
)
# 6. 自动给连写的关键字补空格（字段/中文+关键字粘连自动拆分）
keywords_upper = [
"SELECT", "FROM", "WHERE", "ORDER BY", "GROUP BY",
"AND", "OR", "LIMIT", "DESC", "ASC", "AS",
"INNER JOIN", "LEFT JOIN", "RIGHT JOIN", "ON",
"INSERT INTO", "UPDATE", "SET", "DELETE FROM",
"VALUES", "LIKE", "IN", "BETWEEN", "IS NULL",
"COUNT", "SUM", "AVG", "MAX", "MIN", "OVER"
]
for kw in keywords_upper:
# 字母下划线+关键字粘连拆分，补充第三个参数sql
pattern = r'([a-z_])(' + re.escape(kw) + r')'
sql = re.sub(pattern, r'\1 \2', sql)
# 中文文字+关键字粘连拆分，补充第三个参数sql
pattern_cn = r'([\u4e00-\u9fa5])(' + re.escape(kw) + r')'
sql = re.sub(pattern_cn, r'\1 \2', sql)
# 7. 统一所有SQL关键字大写，格式标准化
keywords_lower = [kw.lower() for kw in keywords_upper]
for kw in keywords_lower:
sql = re.sub(
r'\b' + re.escape(kw) + r'\b',
kw.upper(),
sql,
flags=re.IGNORECASE
)
# 最终再清理一遍多余空格
sql = re.sub(r'\s+', ' ', sql).strip()
return sql
def nl2sql_query(user_input: str) -> dict:
"""
核心业务1：自然语言转SQL、执行查询、AI生成业务总结
对外统一标准返回字典，终端/网页程序均可直接调用，无重复代码
入参：用户自然语言查询需求
返回：包含执行状态、SQL、字段、数据、AI总结、模型原始输出
"""
# 初始化大模型、数据库客户端
llm = get_llm()
db = Mysql80Client()
try:
logger.info("正在生成SQL语句...")
# 1. 加载NL2SQL提示词，填充用户需求传给大模型
prompt = UnifiedPrompt.NL_TO_SQL_PROMPT.format(user_input=user_input)
response = llm.invoke(prompt)
raw_content = response.content.strip()
# 2. 从模型返回文本提取纯净SQL，提取失败直接抛异常
extracted_sql = prompt_helper.extract_sql(raw_content)
if not extracted_sql:
raise Exception("大模型未返回有效SQL，请重新描述需求")
# 3. 清洗SQL修复各类格式问题
clean_sql = clean_sql_spacing(extracted_sql)
logger.info(f"生成SQL：{clean_sql}")
# 4. 数据库执行查询，拿到表头与数据
columns, rows = db.execute_query(clean_sql)
# 5. 如果有数据，调用大模型生成业务解读总结
summary = ""
if rows:
logger.info("正在生成数据总结...")
summary_prompt = f"""
以下是真实的SQL查询结果，请作为电商数据分析师给出简练的业务总结。
SQL语句：{clean_sql}
查询数据：{str(rows)}
重点说明数据反映的业务含义，如有异常值请指出。
"""
summary_resp = llm.invoke(summary_prompt)
summary = summary_resp.content.strip()
# 成功结果返回
return {
"success": True,
"sql": clean_sql,
"columns": columns,
"rows": rows,
"summary": summary,
"raw_llm": raw_content
}
except Exception as e:
# 捕获全流程所有异常，记录日志并返回错误信息
logger.error(f"查询处理失败：{str(e)}")
return {
"success": False,
"error": str(e),
"raw_llm": raw_content if 'raw_content' in dir() else ""
}
finally:
# 无论成功失败，都关闭数据库连接释放资源
db.close()
def sql_tune_analyze(raw_sql: str) -> dict:
"""
核心业务2：SQL性能调优分析
流程：清洗SQL → 获取EXPLAIN执行计划 → AI分析给出优化方案
入参：用户输入待优化SQL
返回：执行状态、清洗后SQL、执行计划字段/内容、调优建议
"""
llm = get_llm()
db = Mysql80Client()
try:
# 先标准化清洗SQL
clean_sql = clean_sql_spacing(raw_sql)
logger.info("正在获取执行计划...")
# 调用数据库封装方法获取EXPLAIN执行计划
columns, plan_rows = db.get_explain_plan(clean_sql)
# 填充调优提示词，传入SQL和执行计划让AI分析瓶颈
logger.info("正在分析性能瓶颈...")
prompt = UnifiedPrompt.SQL_TUNE_PROMPT.format(
sql_input=clean_sql,
explain_data=str(plan_rows)
)
response = llm.invoke(prompt)
return {
"success": True,
"sql": clean_sql,
"plan_columns": columns,
"plan_rows": plan_rows,
"suggestion": response.content.strip()
}
except Exception as e:
logger.error(f"调优分析失败：{str(e)}")
return {
"success": False,
"error": str(e)
}
finally:
# 操作结束关闭数据库连接
db.close()
def main_cli():
"""
终端交互入口主函数
提供循环菜单，支持用户选择查询/调优/退出，纯终端操作
"""
# 程序启动先校验全部配置，失败直接退出菜单
try:
check_config()
except ValueError as e:
print(f"❌ {e}")
return
# 循环交互，不退出可持续多次使用
while True:
print("\n=============== InnoAI SQL 助手 ===============")
print("1. 自然语言生成SQL，自动查询并AI总结数据")
print("2. 输入SQL语句，AI分析执行计划并给出调优方案")
print("0. 退出程序")
choice = input("请输入功能序号: ").strip()
# 功能1：自然语言查数据
if choice == '1':
query = input("请输入你的数据查询需求: ").strip()
if not query:
print("⚠️ 请输入有效需求")
continue
result = nl2sql_query(query)
# 处理失败场景，打印错误与模型原始输出
if not result["success"]:
print(f"\n❌ 处理失败：{result['error']}")
if result.get("raw_llm"):
print(f"大模型原始回复：\n{result['raw_llm']}")
continue
# 成功：打印SQL、格式化表格展示数据、输出业务总结
print(f"\n✅ 生成SQL：")
print(result["sql"])
if result["rows"]:
print(f"\n📊 查询结果（共 {len(result['rows'])} 条）：")
print(tabulate(result["rows"], headers="keys", tablefmt="pretty"))
if result["summary"]:
print(f"\n💡 业务总结：\n{result['summary']}")
else:
print("\n⚠️ 未查询到匹配数据")
# 功能2：SQL性能调优
elif choice == '2':
sql_input = input("\n请输入需要分析的 SQL 语句: ").strip()
if not sql_input:
print("⚠️ 请输入有效SQL")
continue
result = sql_tune_analyze(sql_input)
if not result["success"]:
print(f"\n❌ 分析失败：{result['error']}")
continue
# 打印执行计划表格和AI优化建议
print(f"\n📊 执行计划详情：")
print(tabulate(result["plan_rows"], headers="keys", tablefmt="pretty"))
print(f"\n📈 调优建议：\n{result['suggestion']}")
# 0 退出循环，结束程序
elif choice == '0':
print("程序已安全退出。")
break
# 无效数字输入提示
else:
print("无效输入，请重试。")
print("\n" + "-" * 40)
# 程序入口：直接运行main.py则启动终端菜单
if __name__ == "__main__":
main_cli()

测试

python3 /opt/mysql_ai_tools/main.py