【MySQL + 大模型实战】InnoAI SQL 助手：基于DeepSeek-V4-Pro的电商订单智能查询与性能调优系统

相思难忘成疾

256人浏览 · 2026-07-28 17:43:31

相思难忘成疾 · 2026-07-28 17:43:31 发布

前言

在日常数据库运维中，业务人员想查数据却不会写 SQL，DBA 每天重复处理大量"帮我查个数据"的需求；线上慢查询来了，还要手动跑 EXPLAIN、逐行分析执行计划、再手写索引优化方案——这些重复性劳动占据了大量精力。

能不能让 AI 替我们干这些活？

本文介绍一个完整的实战项目——InnoAI SQL 助手。它以 MySQL 8.0 DBA 运维核心知识点为基础，结合腾讯云 TokenHub 大模型 API，构建了一套电商订单库智能查询与性能调优系统：

（1）自然语言转 SQL：输入"查询用户1001的所有订单"，自动生成合规 SELECT 语句并执行
（2）AI 业务解读：查询结果自动由大模型生成通俗易懂的业务总结
（3）SQL 性能调优：输入任意 SQL，自动解析 EXPLAIN 执行计划，输出标准化索引优化方案
（4）安全加固：仅允许只读查询，拦截一切增删改危险操作

全程无需人工手写、分析 SQL，适配企业轻量化数据查询、SQL 性能优化、AI 智能解读一体化落地场景。

技术栈：openEuler 22.03 SP4 + MySQL 8.0.45 + Python 3.11.9 + LangChain + Streamlit + 腾讯云 TokenHub（Qwen3.5-Plus / DeepSeek-V4-Pro）

一、项目概述

本项目以 MySQL 8.0 DBA 运维核心知识点为基础，结合腾讯云 TokenHub 开源大模型 API，构建一套电商订单库智能查询与性能调优系统。

（1）核心能力

功能	说明
自然语言转 SQL	接收业务自然语言，自动生成合规 MySQL 查询语句
自动执行 + 结构化输出	自动执行查询并输出格式化表格
AI 业务解读	依托大模型完成订单数据业务总结
SQL 性能调优	解析 EXPLAIN 执行计划，输出标准化索引优化方案

（2）核心价值

零门槛数据查询：业务人员无需掌握 SQL 语法，用自然语言即可完成数据查询与报表统计
自动化性能调优：自动解析 EXPLAIN 执行计划，输出可直接执行的索引创建与 SQL 改写方案
安全可控：内置 SQL 安全校验机制，仅允许 SELECT 查询，拦截所有增删改操作
分层解耦设计：代码结构清晰，便于二次开发与教学演示

二、技术架构设计

（1）整体架构分层

系统采用四层架构设计，职责边界清晰：

层级	模块	核心职责
接入交互层	命令行终端 / Streamlit Web	面向用户提供交互入口，支持两种使用模式
Python 程序核心层	main.py / web_main.py / prompts.py / mysql_client.py	业务逻辑调度、提示词管理、数据库封装
底层数据持久层	MySQL 8.0.45	存储电商订单测试数据，作为唯一数据源
AI 大模型服务层	腾讯云 TokenHub	统一 AI 能力出口，支持多模型无缝切换

（2）核心文件说明

文件名	功能定位	核心作用
`main.py`	总调度入口与公共服务层	封装三大核心业务流程，提供命令行交互入口
`web_main.py`	Web 可视化界面层	基于 Streamlit 实现图形化交互，结果渲染
`prompts.py`	提示词工程层	统一管理 NL2SQL 与 SQL 调优 Prompt 模板，SQL 提取工具
`mysql_client.py`	数据访问层	封装 MySQL 连接、查询、执行计划获取，SQL 安全校验
`.env`	独立配置文件	集中存放数据库参数与大模型密钥，敏感信息隔离

三、软硬件环境准备

类别	名称/规格	版本/参数	用途	备注
硬件环境	虚拟机	CPU≥2核，内存≥4GB，磁盘≥20GB	运行 Linux、MySQL、Python 程序	本地虚拟机或云 ECS 均可
软件环境	操作系统	openEuler / RHEL9	项目底层运行系统	需配置外网访问 API
	MySQL	8.0.45	存储电商订单测试业务数据	内置 order_info 订单表
	Python	3.11.9（源码编译）	项目主开发语言	依赖 pymysql、python-dotenv、openai、tabulate
	Shell	Bash	系统环境操作、编译 Python	系统默认自带
网络环境	外网访问	可访问大模型 API	调用大模型接口	防火墙放行 443 端口或关闭
账号资源	腾讯云账号	完成实名认证	获取 API Key，调用大模型	官网：腾讯云产业智变·云启未来 - 腾讯

四、项目环境搭建

（1）系统初始化配置

安装 openEuler 2203_SP4 系统（过程略），完成以下初始化操作：

1、修改主机名

2、关闭SELinux与防火墙

3、配置时间同步

4、安装基础依赖包

#修改主机名
[root@node ~]# hostnamectl set-hostname server

#关闭防火墙
[root@server ~]# systemctl disable --now firewalld
[root@server ~]# systemctl status firewalld
○ firewalld.service - firewalld - dynamic firewall daemon
     Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; ve>
     Active: inactive (dead)
       Docs: man:firewalld(1)

#关闭SELinux
[root@server ~]# sed -i '7s/enforcing/disabled/' /etc/selinux/config
[root@server ~]# reboot
[root@server ~]# getenforce 
Disabled

#配置时间同步服务
[root@server ~]# vim /etc/chrony.conf 
server ntp.aliyun.com iburst
[root@server ~]# systemctl restart chronyd
[root@server ~]# chronyc sources
MS Name/IP address         Stratum Poll Reach LastRx Last sample               
===============================================================================
^* 203.107.6.88                  2   6    17     7  -1434us[-2651us] +/-   30ms

#下载所需软件
[root@server ~]# 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

（2）源码编译安装Python 3.11.9

1、下载源码包

从 Python 官网下载 3.11.9 源码包，上传至/usr/local/src目录

下载地址：Python 3.11.9https://www.python.org/downloads/release/python-3119/

2、解压并编译

解压：

[root@server src]# tar -xzvf Python-3.11.9.tgz

编译配置：

[root@server src]# cd Python-3.11.9
[root@server Python-3.11.9]# ./configure --prefix=/usr/local/python3.11 --enable-shared

多核编译安装：

[root@server Python-3.11.9]# make -j$(nproc) && make install

3、配置动态链接库

[root@server Python-3.11.9]# echo "/usr/local/python3.11/lib" > /etc/ld.so.conf.d/python311.conf
[root@server Python-3.11.9]# ldconfig

目的：解决libpython缺失报错

4、建立全局软连接

[root@server Python-3.11.9]# ln -s /usr/local/python3.11/bin/python3.11 /usr/local/bin/python3
[root@server Python-3.11.9]# ln -s /usr/local/python3.11/bin/pip3.11 /usr/local/bin/pip3

5、验证安装

#验证Python安装
[root@server Python-3.11.9]# python3 -V
Python 3.11.9
[root@server Python-3.11.9]# pip3 -V
pip 24.0 from /usr/local/python3.11/lib/python3.11/site-packages/pip (python 3.11)

#校验SSL模块
[root@server Python-3.11.9]# python3 -c "import ssl; print(ssl.OPENSSL_VERSION)"

（3）配置 pip 源与安装项目依赖

1. 配置阿里镜像源

[root@server ~]# mkdir .pip
[root@server ~]# vim ~/.pip/pip.conf
[root@server ~]# cat ~/.pip/pip.conf 
[global]
index-url = http://mirrors.aliyun.com/pypi/simple/

[install]
trusted-host=mirrors.aliyun.com

2、安装依赖

[root@server ~]# pip3 install --upgrade pip
[root@server ~]# pip3 install pymysql python-dotenv tabulate langchain langchain-openai

（4）部署并初始化 MySQL 8.0.45

1、虚拟机部署部分的超详细版（含原理与高可用分析）见本人专栏文章

告别RPM包！生产环境MySQL通用二进制包安装与高可用配置全解析文章浏览阅读274次，点赞10次，收藏5次。本文详细介绍了在OpenEuler 22.03 LTS系统上使用通用二进制包安装MySQL 8.0.45的完整流程。首先阐述了二进制包安装的优势，包括环境解耦、版本可控等特点。随后逐步演示了软件获取、解压安装、用户权限配置、数据库初始化、启动验证等关键步骤，并解决了依赖库报错问题。最后配置了systemd服务和环境变量，使MySQL服务可方便管理。文章还指出了当前配置的不足，如性能参数未优化、高可用机制缺失等，为后续生产环境部署提供了改进方向。整个安装过程注重标准化和安全规范，为数据库运维打下良好基础。_glibc 二进制mysql 可用于生产环境吗?https://blog.csdn.net/2502_90206768/article/details/162911810?spm=1011.2124.3001.6209

2、创建测试库与订单表

mysql> 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='电商订单业务表';
Query OK, 0 rows affected (0.01 sec)


mysql> 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');
Query OK, 20 rows affected (0.00 sec)
Records: 20  Duplicates: 0  Warnings: 0

五、获取大模型 API Key

（1）访问腾讯云官网，注册账号并完成实名认证

腾讯云产业智变·云启未来 - 腾讯腾讯云(tencent cloud)为数百万的企业和开发者提供安全稳定的云计算服务，涵盖云服务器、云数据库、云存储、视频与CDN、域名注册等全方位云服务和各行业解决方案。https://cloud.tencent.com/

（2）进入 TokenHub 控制台，进入 "API Key 管理" → "新建 API 密钥"

（3）填写密钥名称并保存，记录生成的 API Key

（4）模型选择：推荐使用`qwen3.5-plus`或`deepseek-v4-pro`

六、核心脚本开发

（1）新建目录及脚本

[root@server ~]# mkdir -p /opt/mysql_ai_tools
[root@server ~]# cd /opt/mysql_ai_tools
[root@server mysql_ai_tools]# touch main.py mysql_client.py prompts.py web_main.py .env
[root@server mysql_ai_tools]# tree -La 1
.
|-- .env
|-- main.py
|-- mysql_client.py
|-- prompts.py
|-- web_main.py
`-- \343.env

0 directories, 6 files
[root@server mysql_ai_tools]#

（2）编写环境变量配置（.env）

[root@server mysql_ai_tools]# 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=你自己的API密钥

LLM_BASE_URL=https://tokenhub.tencentmaas.com/v1

LLM_MODEL_NAME=DeepSeek-V4-Pro

LLM_TEMPERATURE=0

加固文件权限：

[root@server mysql_ai_tools]# chmod 600 /opt/mysql_ai_tools/.env

（3）编写数据库连接脚本（mysql_client.py）

[root@server mysql_ai_tools]# vim mysql_client.py

# -*- coding: utf-8 -*-
# 文件名: mysql_client.py
# 功能: MySQL8.0数据库统一封装类
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:
        """SQL安全校验，仅允许SELECT查询"""
        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查询"""
        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执行计划"""
        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()

（4）编写提示词工程脚本（prompts.py）

[root@server mysql_ai_tools]# vim prompts.py 

# -*- coding: utf-8 -*-
# 文件名: prompts.py
# 功能: 统一管理大模型提示词模板与SQL提取工具
import re

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

    # 自然语言转SQL提示词
    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. 中文别名内部不能带空格。
【用户需求】
{{user_input}}
"""

    # SQL性能调优提示词
    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语句"""
        if not response_text:
            return ""
        
        # 优先级1: markdown sql代码块
        match = re.search(
            r"```sql\s*(.*?)\s*```", 
            response_text, 
            re.DOTALL | re.IGNORECASE
        )
        if match:
            return match.group(1).strip()
        
        # 优先级2: <sql>标签
        match = re.search(
            r"<sql>\s*(.*?)\s*</sql>", 
            response_text, 
            re.DOTALL | re.IGNORECASE
        )
        if match:
            return match.group(1).strip()
        
        # 优先级3: SELECT开头兜底匹配
        match = re.search(
            r"(SELECT\s+.*?;)", 
            response_text, 
            re.DOTALL | re.IGNORECASE
        )
        if match:
            return match.group(1).strip()
        
        return ""

# 全局单例
prompt_helper = UnifiedPrompt()

（5）编写程序入口脚本（main.py）

[root@server mysql_ai_tools]# vim main.py 

# -*- coding: utf-8 -*-
# 文件名: 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:
    """启动前配置校验"""
    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:
    """初始化大模型客户端"""
    api_key = os.getenv("LLM_API_KEY")
    base_url = os.getenv("LLM_BASE_URL")
    model_name = os.getenv("LLM_MODEL_NAME")
    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标准化清洗函数"""
    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. 中文标点转英文
    sql = sql.replace('，', ',').replace('；', ';').replace('（', '(').replace('）', ')')
    
    # 4. 合并连续空格
    sql = re.sub(r'\s+', ' ', sql).strip()
    
    # 5. 清理别名中的空格
    def _clean_alias_space(match):
        prefix = match.group(1)
        alias = match.group(2)
        alias_clean = re.sub(r'\s+', '', alias)
        return f"{prefix} {alias_clean}"
    
    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",
        "LIKE", "IN", "BETWEEN", "IS NULL",
        "COUNT", "SUM", "AVG", "MAX", "MIN"
    ]
    for kw in keywords_upper:
        pattern = r'([a-z_])(' + re.escape(kw) + r')'
        sql = re.sub(pattern, r'\1 \2', sql)
        pattern_cn = r'([\u4e00-\u9fa5])(' + re.escape(kw) + r')'
        sql = re.sub(pattern_cn, r'\1 \2', sql)
    
    # 7. 关键字统一大写
    for kw in keywords_upper:
        sql = re.sub(
            r'\b' + re.escape(kw.lower()) + 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并执行查询"""
    llm = get_llm()
    db = Mysql80Client()
    try:
        logger.info("正在生成SQL语句...")
        prompt = UnifiedPrompt.NL_TO_SQL_PROMPT.format(user_input=user_input)
        response = llm.invoke(prompt)
        raw_content = response.content.strip()
        
        extracted_sql = prompt_helper.extract_sql(raw_content)
        if not extracted_sql:
            raise Exception("大模型未返回有效SQL,请重新描述需求")
        
        clean_sql = clean_sql_spacing(extracted_sql)
        logger.info(f"生成SQL:{clean_sql}")
        
        columns, rows = db.execute_query(clean_sql)
        
        summary = ""
        if rows:
            logger.info("正在生成数据总结...")
            summary_prompt = f"""
以下是真实的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性能调优分析"""
    llm = get_llm()
    db = Mysql80Client()
    try:
        clean_sql = clean_sql_spacing(raw_sql)
        logger.info("正在获取执行计划...")
        columns, plan_rows = db.get_explain_plan(clean_sql)
        
        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()
        
        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
            
            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⚠️ 未查询到匹配数据")
        
        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
            
            print(f"\n📊 执行计划详情:")
            print(tabulate(result["plan_rows"], headers="keys", tablefmt="pretty"))
            print(f"\n📈 调优建议:\n{result['suggestion']}")
        
        elif choice == '0':
            print("程序已安全退出。")
            break
        
        else:
            print("无效输入,请重试。")
        print("\n" + "-" * 40)

if __name__ == "__main__":
    main_cli()

七、功能测试

（1）自然语言查询测试

启动程序：

[root@server mysql_ai_tools]# python3 /opt/mysql_ai_tools/main.py

（2）SQL 性能调优测试

输入待分析 SQL：

SELECT order_name AS 商品名称, pay_amount AS 支付金额, create_time AS 下单时间 FROM order_info WHERE user_id = 1001;

八、Web 可视化版本（Streamlit）

（1）Streamlit 介绍

Streamlit 是开源、纯 Python Web 开发库，核心定位：不用写 HTML/CSS/JS，只用 Python 就能快速做交互式网页。

1、特点：

MIT 开源，商用免费
组件开箱即用，内置全套交互控件
开发热重载，改代码自动刷新页面
部署极其简单，支持服务器 systemd 常驻
无缝兼容 Python 全生态

2、局限性：

并发性能一般，单进程模型
自定义 UI 能力弱
原生无登录权限控制

（2）部署流程

1、安装并检查 Streamlit 依赖

[root@server mysql_ai_tools]# /usr/local/python3.11/bin/pip3 install streamlit

[root@server mysql_ai_tools]# /usr/local/python3.11/bin/python3 -c "import langchain_openai; print('3.11依赖全部正常')"
3.11依赖全部正常

2、创建 Streamlit 可视化入口（web_main.py）

[root@server mysql_ai_tools]# vim web_main.py 

# -*- coding: utf-8 -*-
# 文件名: web_main.py
# 功能: Streamlit网页可视化界面
import streamlit as st
import os
import re
from main import check_config, nl2sql_query, sql_tune_analyze

st.set_page_config(page_title="InnoAI SQL 助手", layout="wide")

# 全局样式优化
st.markdown("""
<style>
html, body, [class*="css"] {
    font-size: 15px;
    line-height: 1.6;
    font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif;
}
.block-container {
    padding-top: 2.5rem;
    padding-bottom: 3rem;
    max-width: 1100px;
    margin: 0 auto;
}
h1 {
    font-size: 2rem !important;
    font-weight: 600;
    margin-bottom: 1.8rem;
}
h2, h3 {
    font-weight: 600;
    margin-top: 1.5rem;
    margin-bottom: 1rem;
}
.stButton > button {
    font-size: 15px !important;
    font-weight: 500;
    padding: 0.65rem 2.2rem !important;
    border-radius: 8px !important;
    min-width: 160px;
}
.stCodeBlock {
    position: relative !important;
    border-radius: 8px !important;
}
.stCodeBlock code, .stCodeBlock pre {
    font-size: 14.5px !important;
    line-height: 1.7 !important;
    font-family: "JetBrains Mono", Consolas, Monaco, monospace;
}
.stDataFrame {
    font-size: 14.5px;
    border-radius: 8px;
    overflow: hidden;
}
</style>
""", unsafe_allow_html=True)

# 配置校验
if "config_checked" not in st.session_state:
    try:
        check_config()
        st.session_state.config_checked = True
    except ValueError as e:
        st.error(f"配置错误:{e}")
        st.stop()

# 侧边栏导航
with st.sidebar:
    st.title("🚀 InnoAI SQL 助手")
    st.info(f"当前模型:{os.getenv('LLM_MODEL_NAME', '未知')}")
    page = st.radio("功能导航", ["🔍 数据查询与总结", "⚙️ SQL 性能调优"])

# 页面1: 自然语言查询
if page == "🔍 数据查询与总结":
    st.header("💬 自然语言转 SQL 查询")
    user_input = st.text_area("请输入你的业务查询需求:", height=150)
    
    if st.button("🚀 生成并执行", type="primary"):
        input_trim = user_input.strip()
        if not input_trim:
            st.warning("请输入有效的业务查询需求后再提交")
        elif re.match(r'(?i)^\s*SELECT\s+', input_trim):
            st.warning("此处请输入自然语言描述的查询需求,请勿直接粘贴 SQL 语句。")
        else:
            with st.spinner("AI 正在生成 SQL 并查询数据..."):
                result = nl2sql_query(user_input)
                if not result["success"]:
                    st.error(f"处理失败:{result['error']}")
                    if result.get("raw_llm"):
                        with st.expander("查看大模型原始回复"):
                            st.code(result["raw_llm"])
                else:
                    st.success("SQL 生成并执行成功")
                    st.code(result["sql"], language="sql")
                    if result["rows"]:
                        st.dataframe(result["rows"], use_container_width=True)
                        if result["summary"]:
                            st.markdown("### 💡 业务总结")
                            st.info(result["summary"])
                    else:
                        st.info("未查询到匹配的数据")

# 页面2: SQL性能调优
else:
    st.header("🔧 SQL 性能调优分析")
    raw_sql = st.text_area("请输入待分析的 SQL 语句:", height=200)
    
    if st.button("📊 开始分析", type="primary"):
        input_trim = raw_sql.strip()
        if not input_trim:
            st.warning("请输入有效的 SQL 语句后再提交")
        elif not re.match(r'(?i)^\s*SELECT\s+', input_trim):
            st.warning("此处请输入待分析的 SELECT SQL 语句,请勿输入自然语言描述。")
        else:
            with st.spinner("正在获取执行计划并分析..."):
                result = sql_tune_analyze(raw_sql)
                if not result["success"]:
                    st.error(f"分析失败:{result['error']}")
                else:
                    st.success("执行计划获取成功")
                    st.dataframe(result["plan_rows"], use_container_width=True)
                    st.markdown("### 💡 调优建议")
                    st.markdown(result["suggestion"])

3、配置 systemd 后台服务

[root@server mysql_ai_tools]# vim /etc/systemd/system/mysql-ai-web.service

[Unit]
Description=InnoAI SQL Streamlit Web Tool
After=network.target mysqld.service

[Service]
Type=simple
User=root
WorkingDirectory=/opt/mysql_ai_tools
ExecStart=/usr/local/python3.11/bin/python3 -m streamlit run web_main.py \
  --server.address 0.0.0.0 \
  --server.port 8501 \
  --server.headless true
Restart=always
RestartSec=3
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

启动服务：

[root@server mysql_ai_tools]# vim /etc/systemd/system/mysql-ai-web.service
[root@server mysql_ai_tools]# systemctl daemon-reload
[root@server mysql_ai_tools]# systemctl enable --now mysql-ai-web
Created symlink /etc/systemd/system/multi-user.target.wants/mysql-ai-web.service → /etc/systemd/system/mysql-ai-web.service.

4、访问测试

即可看到 InnoAI SQL 助手的 Web 可视化界面，支持自然语言查询和 SQL 性能调优两大功能。

九、总结

本文完整实现了 InnoAI SQL 助手 项目，从 openEuler 系统初始化、Python 3.11.9 源码编译、MySQL 8.0.45 部署，到大模型 API 对接、Python 分层脚本开发、功能测试、Streamlit Web 可视化部署，全流程可复现。

亮点	说明
🔒 安全设计	SQL 黑名单拦截 + .env 密钥隔离 + 600 权限加固
🧠 提示词工程	严格约束模型输出格式，三层 SQL 提取兜底
🛡️ 容错清洗	7 步 SQL 标准化清洗，兼容各大模型输出差异
📐 分层架构	数据层/提示词层/业务层/展示层完全解耦
🌐 双端支持	CLI 终端 + Streamlit Web 可视化，一套核心逻辑

快递鸟一站式物流API解决方案

电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。

更多推荐

星宇智算 AI 视频工作台 AI 画布：电商商用视觉一站式 AI 生成平台落地解析

快递鸟社区

多语言商城系统标准开发与跨境电商平台架构规范

本文系统阐述了多语言跨境电商商城的开发规范与架构设计。核心采用微服务架构分层设计，包含网关层、业务服务层和支撑服务层。重点构建多语言本地化架构，通过"主表+翻译表"数据模型、统一词条库和区域感知实现全球化支持。在开发规范方面，制定了API设计、错误处理、安全合规等标准，特别强调GDPR等区域合规要求。关键模块开发涵盖商品中心多语言管理、订单风控和支付结算等核心功能。最后提出容器