一、项目简介

本项目基于国产 openEuler 操作系统、MySQL 8.0 数据库，结合腾讯云 TokenHub 大模型网关，开发轻量化 InnoAI SQL 智能助手系统，面向电商订单数据业务场景，支持接收自然语言需求自动生成合规只读 SQL 查询语句、执行数据统计并输出 AI 业务分析结果，同时可解析 SQL 执行计划、智能生成索引优化方案。项目单人完成开发与部署，提供命令行终端、Web 网页两套交互方式，无需使用者掌握专业 SQL 开发技能，能够替代人工制作统计报表、手动分析 SQL 性能瓶颈等重复运维工作，实现数据查询、SQL 调优、AI 智能数据分析。

核心实现功能：

1. 通过中文自然语言描述业务需求，自动生成只读 SQL，执行查询并输出 AI 业务分析总结；
2. 支持传入 SQL 语句，自动获取执行计划，识别性能瓶颈，输出索引优化方案；
3. 内置安全防护机制，拦截增删改、建删表等高危 SQL；敏感配置独立存储，避免密钥泄露。

二、整体逻辑说明

整套工具采用分层解耦思路开发，模块职责划分清晰，便于后续迭代拓展：

1. 交互入口：分为服务器终端交互式菜单、Streamlit 可视化网页，两端复用同一套底层逻辑，减少重复开发；
2. Python 核心业务层：包含数据库封装模块、大模型提示词模块、程序调度入口、网页 UI 代码、独立配置文件；
3. 数据存储层：openEuler 虚拟机内部署 MySQL8.0，创建电商订单测试表，使用模拟订单数据开展测试；
4. AI 能力支撑：依托 TokenHub 大模型聚合网关，兼容多款主流大模型，统一接口调用。

三、环境部署步骤

1. openEuler 系统初始化

关闭防火墙与 SELinux 防止端口拦截，配置阿里云时间同步服务，提前安装 gcc、openssl 等编译依赖包，查看服务器内网 IP 命令：hostname -I。

2. 源码编译 Python3.11.9

从官网 https://www.python.org/downloads/release/python-3119/ 下载 Python 源码包上传服务器编译安装，配置国内 pip 镜像加速依赖下载，安装 pymysql、streamlit、langchain 等项目所需第三方库。

3. MySQL8.0 单机部署

下载 https://downloads.mysql.com/archives/community/ 采用二进制包方式部署 MySQL，创建专用系统用户降低权限风险；新建 order_info 订单业务测试表，导入 20 条模拟订单数据。

4. 大模型平台密钥配置

注册腾讯云并完成实名认证，在 TokenHub 平台创建 API 密钥；优先选择无冗余思考链输出的模型，避免多余文本干扰程序提取 SQL。

四、项目核心文件说明

项目根目录一共 5 个核心代码文件：

文件	说明
.env	存放数据库、腾讯云 TokenHub 大模型密钥等配置，分离敏感信息与业务代码
mysql_client.py	封装 MySQL 连接、查询与执行计划获取，拦截危险 SQL 保障数据库安全
prompts.py	统一存放生成 SQL、调优分析两套 AI 提示词，并提供提取纯净 SQL 的工具函数
main.py	项目总调度，串联数据库、大模型、提示词，提供终端菜单交互，实现完整业务逻辑
web_main.py	Streamlit 可视化网页界面，做输入校验、图形化展示结果，复用 main 核心逻辑

4.1 提示词工程层（prompts.py）

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 ""

        # 匹配markdown代码块
        match = re.search(r"```sql\s*(.*?)\s*```", response_text, re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()

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

        # 兜底匹配SELECT开头
        match = re.search(r"(SELECT\s+.*?;)", response_text, re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()

        return ""

prompt_helper = UnifiedPrompt()

4.2 数据库封装层（mysql_client.py）

import pymysql
from pymysql.cursors import DictCursor
import os
import logging

logger = logging.getLogger(__name__)

class Mysql80Client:
    """MySQL 8.0 客户端封装"""

    # 高危关键字黑名单
    DANGER_KEYWORDS = [
        'INSERT', 'UPDATE', 'DELETE', 'DROP', 'ALTER',
        'TRUNCATE', 'CREATE', 'REPLACE', 'GRANT', 'REVOKE'
    ]

    def __init__(self):
        self.connection = None
        self._connect()

    def _connect(self):
        """建立数据库连接"""
        self.connection = pymysql.connect(
            host=os.getenv("MYSQL_HOST", "localhost"),
            port=int(os.getenv("MYSQL_PORT", 3306)),
            user=os.getenv("MYSQL_USER", "root"),
            password=os.getenv("MYSQL_PASSWORD", ""),
            database=os.getenv("MYSQL_DB", ""),
            charset='utf8mb4',
            cursorclass=DictCursor
        )

    def _safety_check(self, sql: str) -> bool:
        """SQL安全校验，只允许SELECT查询"""
        sql_upper = sql.upper()
        for keyword in self.DANGER_KEYWORDS:
            if keyword in sql_upper:
                return False
        if not sql_upper.strip().startswith("SELECT"):
            return False
        return True

    def execute_query(self, sql: str) -> tuple:
        """执行查询，返回(字段列表, 数据行)"""
        if not self._safety_check(sql):
            raise ValueError("检测到危险SQL语句，已拒绝执行")
        with self.connection.cursor() as cursor:
            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
        return columns, rows

    def get_explain_plan(self, sql: str) -> tuple:
        """获取EXPLAIN执行计划"""
        with self.connection.cursor() as cursor:
            cursor.execute(f"EXPLAIN {sql}")
            columns = [desc[0] for desc in cursor.description]
            plan_rows = cursor.fetchall()
        return columns, plan_rows

    def close(self):
        """关闭连接"""
        if self.connection:
            self.connection.close()

4.3 核心业务调度层（main.py）

import os
import logging
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from mysql_client import Mysql80Client
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:
    """初始化大模型客户端"""
    return ChatOpenAI(
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
        model=os.getenv("LLM_MODEL_NAME"),
        temperature=float(os.getenv("LLM_TEMPERATURE", 0.1))
    )


def nl2sql_query(user_input: str) -> dict:
    """自然语言转SQL、执行查询、生成业务总结"""
    llm = get_llm()
    db = Mysql80Client()

    try:
        logger.info("正在生成SQL语句...")

        # 调用大模型生成SQL
        prompt = UnifiedPrompt.NL_TO_SQL_PROMPT.format(user_input=user_input)
        response = llm.invoke(prompt)
        raw_content = response.content.strip()

        # 提取SQL
        extracted_sql = prompt_helper.extract_sql(raw_content)
        if not extracted_sql:
            raise Exception("大模型未返回有效SQL，请重新描述需求")

        logger.info(f"生成SQL：{extracted_sql}")

        # 执行查询
        columns, rows = db.execute_query(extracted_sql)

        # 生成业务总结
        summary = ""
        if rows:
            logger.info("正在生成数据总结...")
            summary_prompt = f"""
以下是真实的SQL查询结果，请作为电商数据分析师给出简练的业务总结。
SQL语句：{extracted_sql}
查询数据：{str(rows)}
重点说明数据反映的业务含义，如有异常值请指出。
"""
            summary_resp = llm.invoke(summary_prompt)
            summary = summary_resp.content.strip()

        return {
            "success": True,
            "sql": extracted_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:
    """SQL性能调优分析"""
    llm = get_llm()
    db = Mysql80Client()

    try:
        logger.info("正在获取执行计划...")

        # 获取执行计划
        columns, plan_rows = db.get_explain_plan(raw_sql)

        # 调用大模型分析
        logger.info("正在分析性能瓶颈...")
        prompt = UnifiedPrompt.SQL_TUNE_PROMPT.format(
            sql_input=raw_sql,
            explain_data=str(plan_rows)
        )
        response = llm.invoke(prompt)

        return {
            "success": True,
            "sql": raw_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()

4.4 Streamlit 可视化层（web_main.py）

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")

# 配置校验（会话缓存）
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"])

五、业务 SQL 功能测试

设计多组真实电商业务用例完成全量测试，覆盖分组统计、区间筛选、模糊匹配、聚合计算等场景：

自然语言查询测试

输入需求：统计 1001、1002、1003 每个用户的订单总消费金额与订单笔数，按总消费从高到低排序。

工具自动生成规范 SELECT 语句，执行后输出格式化表格，同时 AI 输出分层用户价值、复购行为等业务分析总结。

SQL 性能调优测试

将上一步生成的统计 SQL 粘贴至调优功能，程序自动抓取 EXPLAIN 执行计划，识别全表扫描、临时表、文件排序等性能问题，输出可直接执行的覆盖索引创建语句，同时给出优化后的标准 SQL。

边界场景测试

针对无匹配数据、空白输入、超长复合需求、含特殊字符的商品名称等场景验证，程序不会崩溃报错；安全测试验证工具不会生成 DELETE、ALTER 等危险修改语句。

六、功能实测演示

• 终端自然语言查询：运行 python3 main.py 输入中文需求，自动生成 SQL、打印表格并附带业务总结；
• 终端 SQL 性能调优：粘贴查询语句，自动获取执行计划，输出索引优化方案；
• Web 网页访问：浏览器输入 服务器IP:8501 可视化操作，业务人员可自助查询、调优 SQL。

七、项目开发难点与解决方案

1. 大模型输出格式杂乱：通过标准化提示词约束输出，搭配正则函数精准提取 SQL；
2. 数据安全风险：增加 SQL 黑白名单校验，仅放行 SELECT 只读查询；
3. Web 服务易中断：编写 systemd 托管程序，实现后台常驻、开机自启。

八、项目优势与学习收获

1. 项目优势

适配国产 openEuler 系统，轻量化单机部署，低配虚拟机即可运行；Streamlit 零前端开发，快速交付可视化网页；完善安全机制，密钥与代码分离，杜绝误删数据、信息泄露；自动化完成报表统计、慢 SQL 分析，减少开发与运维重复工作。

2. 学习收获

完整实践国产 Linux 服务部署、Python 模块化开发、大模型 API 调用、MySQL 性能优化知识，掌握分层开发、进程托管、程序安全校验等工程化开发思路。