一、项目实施全流程1.1openEuler 系统初始化配置1.1.1 系统安全与网络优化刚装好 openEuler 系统防火墙、SELinux 会拦截端口访问时间同步错乱先做系统基础优化。1.1.2 源码编译 Python3.11.91.下载源码包上传至/usr/local/src解压编译系统自带 Python3 版本过低选择源码编译全新 Python1.1.3配置国内阿里 pip 源国外官方 pip 源下载依赖速度极慢切换阿里镜像站加速下载2.2 MySQL8.0.45 部署 测试订单库初始化2.1.1 创建订单测试库、表、20 条测试数据2.2 腾讯云 TokenHub 大模型平台准备1.进入 TokenHub 平台 - API Key 管理新建 API 密钥复制保存 sk 开头的密钥2.模型广场选择deepseek-v4-pro二、项目 Python 源码开发逐文件带功能 设计思路讲解3.1 创建项目目录3.2编写环境变量脚本3.3 编写数据库连接脚本(vim /opt/mysql_ai_tools/mysql_client.py)# -*- coding: utf-8 -*-# 文件名mysql_client.py# 功能MySQL8.0数据库统一封装类# 作用封装数据库连接、普通查询、EXPLAIN执行计划、SQL安全拦截统一抛出友好异常给上层业务调用import pymysqlimport osimport refrom dotenv import load_dotenv# 加载项目根目录下.env文件的数据库配置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(hostself.host,portself.port,userself.user,passwordself.password,databaseself.database,charsetutf8mb4, # 支持中文、emoji完整字符集cursorclasspymysql.cursors.DictCursor # 查询结果以字典返回方便按字段取值)except pymysql.MySQLError as e:# MySQL专属连接错误提示账号、地址、密码排查方向raise Exception(f数据库连接失败请检查地址/账号/密码{e.args[1]})except Exception as e:# 其余未知连接异常统一捕获raise Exception(f数据库连接异常{str(e)})staticmethoddef _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: (字段名列表, 全部数据行字典列表)# 执行SQL前先做安全校验拦截危险语句self._check_sql_safety(sql)try:# with自动管理游标用完自动释放资源with self.conn.cursor() as cursor:cursor.execute(sql)# 提取查询结果表头字段columns [desc[0] for desc in cursor.description]# 读取全部查询数据rows cursor.fetchall()return columns, rowsexcept pymysql.MySQLError as e:# 捕获SQL语法、表不存在等数据库执行错误raise Exception(fSQL执行失败错误码 {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: (执行计划表头, 执行计划详情数据)# 同样先校验SQL安全性self._check_sql_safety(sql)# 拼接EXPLAIN关键字生成分析语句explain_sql fEXPLAIN {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, rowsexcept 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()3.4. 编写sql转换脚本(vim /opt/mysql_ai_tools/prompts.py)# -*- coding: utf-8 -*-# 文件名prompts.py# 功能统一管理项目全部大模型提示词模板附带SQL提取工具静态方法# 作用把AI提示词和业务代码解耦统一约束模型输出格式降低SQL解析报错概率import reclass UnifiedPrompt:提示词统一管理类优势所有SQL生成、性能分析提示词集中存放表结构仅维护一处修改不用多处同步通过严格规则约束大模型输出减少格式错乱、编造字段、危险SQL等幻觉问题# 全局共用数据表结构 # 只在此维护订单表字段下方两套提示词会自动引用改表结构只需改这里一处TABLE_SCHEMA 表名: order_info (订单信息表)字段说明:- id: 订单ID (主键INT类型)- user_id: 用户ID (INT类型)- order_name: 商品名称 (VARCHAR类型)- pay_amount: 支付金额 (DECIMAL类型)- create_time: 下单时间 (DATETIME类型)# 模板1自然语言转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. 中文别名内部不能带空格例订单ID正确、订单 ID错误避免数据库语法报错。【用户需求】{{user_input}}# 模板2SQL性能调优分析专用提示词 SQL_TUNE_PROMPT f你是资深 MySQL DBA 性能优化专家。【任务目标】根据原始SQL EXPLAIN执行计划数据定位查询性能问题并给出可直接落地的优化方案。【表结构参考】{TABLE_SCHEMA}【待分析SQL】{{sql_input}}【执行计划数据】{{explain_data}}【输出要求】1. 先点明核心性能问题全表扫描、无索引、索引失效、扫描行数过多等。2. 给出完整建索引SQL语句可直接复制执行。3. 若原SQL写法存在缺陷提供改写后的完整优化SQL。4. 内容简洁、分点罗列不输出多余废话便于用户快速阅读。staticmethoddef extract_sql(response_text: str) - str:静态工具方法从大模型返回的完整文本里剥离出纯净SQL语句三层匹配优先级兼容不同大模型的输出格式提升提取成功率:param response_text: 大模型原始完整返回内容:return: 清洗后的纯SQL字符串提取失败返回空字符串# 空文本直接返回if not response_text:return # 优先级1匹配最标准markdown sql代码块项目提示词强制要求的格式match re.search(rsql\s*(.*?)\s*, response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 优先级2兼容自定义sql标签格式备用兼容方案match re.search(rsql\s*(.*?)\s*/sql, response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 优先级3兜底匹配直接抓取以SELECT开头、分号结尾的SQL片段match re.search(r(SELECT\s.*?;), response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 三层规则全部匹配不到说明无有效SQL返回空return # 全局单例实例外部文件导入后直接调用 prompt_helper.方法名无需重复实例化prompt_helper UnifiedPrompt()3.5. 编写程序入口脚本(vim /opt/mysql_ai_tools/main.py)# -*- coding: utf-8 -*-# 文件名main.py# 功能项目核心业务逻辑 终端交互式菜单入口# 作用统一封装大模型调用、SQL清洗、数据库交互两大核心业务命令行/网页共用底层函数import osimport reimport loggingfrom dotenv import load_dotenv# 兼容OpenAI标准大模型接口适配腾讯云TokenHubfrom langchain_openai import ChatOpenAI# 导入数据库操作封装类from mysql_client import Mysql80Client# 表格格式化打印工具美化终端输出查询结果from tabulate import tabulate# 导入提示词管理类与全局实例from prompts import UnifiedPrompt, prompt_helper# 加载.env文件里所有数据库、大模型配置load_dotenv()# 全局日志配置替代print记录运行时间、日志级别、报错信息方便排障logging.basicConfig(levellogging.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_keyapi_key,base_urlbase_url,modelmodel_name,temperaturetemperature)def clean_sql_spacing(sql: str) - str:SQL标准化清洗工具函数兜底修复各大模型输出格式解决中文空格别名、中文标点、特殊空白、关键字连写等语法报错问题入参大模型原始SQL字符串返回清洗后可直接执行的标准英文SQLif 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*|\sFROM\b|\sWHERE\b|\sORDER\b|\sGROUP\b|\sLIMIT\b|\s*;),_clean_alias_space,sql,flagsre.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:# 字母下划线关键字粘连拆分补充第三个参数sqlpattern r([a-z_])( re.escape(kw) r)sql re.sub(pattern, r\1 \2, sql)# 中文文字关键字粘连拆分补充第三个参数sqlpattern_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,flagsre.IGNORECASE)# 最终再清理一遍多余空格sql re.sub(r\s, , sql).strip()return sqldef 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_inputuser_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:核心业务2SQL性能调优分析流程清洗SQL → 获取EXPLAIN执行计划 → AI分析给出优化方案入参用户输入待优化SQL返回执行状态、清洗后SQL、执行计划字段/内容、调优建议llm get_llm()db Mysql80Client()try:# 先标准化清洗SQLclean_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_inputclean_sql,explain_datastr(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(⚠️ 请输入有效需求)continueresult 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], headerskeys, tablefmtpretty))if result[summary]:print(f\n 业务总结\n{result[summary]})else:print(\n⚠️ 未查询到匹配数据)# 功能2SQL性能调优elif choice 2:sql_input input(\n请输入需要分析的 SQL 语句: ).strip()if not sql_input:print(⚠️ 请输入有效SQL)continueresult sql_tune_analyze(sql_input)if not result[success]:print(f\n❌ 分析失败{result[error]})continue# 打印执行计划表格和AI优化建议print(f\n 执行计划详情)print(tabulate(result[plan_rows], headerskeys, tablefmtpretty))print(f\n 调优建议\n{result[suggestion]})# 0 退出循环结束程序elif choice 0:print(程序已安全退出。)break# 无效数字输入提示else:print(无效输入请重试。)print(\n - * 40)# 程序入口直接运行main.py则启动终端菜单if __name__ __main__:main_cli()三、测试1.查询用户 1001 的所有订单展示商品名称、支付金额和下单时间2.SELECT user_id AS 用户ID, SUM(pay_amount) AS 总消费金额, COUNT(id) AS 订单笔数 FROM order_info WHERE user_id IN (1001, 1002, 1003) GROUP BY user_idORDER BY 总消费金额 DESC四、web可视化版本4.1 web_main.py Streamlit 可视化网页功能说明基于 Streamlit 实现可视化网页界面复用 main.py 封装好的业务函数提供浏览器远程操作入口为什么这么设计1.纯 Python 开发网页零基础快速实现可视化页面作为项目拓展加分功能2.页面做输入前置校验区分自然语言输入框和 SQL 输入框防止用户操作混淆3.美化页面样式表格、代码块、按钮优化演示项目时观感更好4.支持展开面板查看大模型原始返回方便调试排错。4.2安装 Python 依赖4.3创建 Streamlit 可视化入口 web_main.pyvim /opt/mysql_ai_tools/web_main.py# -*- coding: utf-8 -*-# 文件名main.py# 功能项目核心业务逻辑 终端交互式菜单入口# 作用统一封装大模型调用、SQL清洗、数据库交互两大核心业务命令行/网页共用底层函数import osimport reimport loggingfrom dotenv import load_dotenv# 兼容OpenAI标准大模型接口适配腾讯云TokenHubfrom langchain_openai import ChatOpenAI# 导入数据库操作封装类from mysql_client import Mysql80Client# 表格格式化打印工具美化终端输出查询结果from tabulate import tabulate# 导入提示词管理类与全局实例from prompts import UnifiedPrompt, prompt_helper# 加载.env文件里所有数据库、大模型配置load_dotenv()# 全局日志配置替代print记录运行时间、日志级别、报错信息方便排障logging.basicConfig(levellogging.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_keyapi_key,base_urlbase_url,modelmodel_name,temperaturetemperature)def clean_sql_spacing(sql: str) - str:SQL标准化清洗工具函数兜底修复各大模型输出格式解决中文空格别名、中文标点、特殊空白、关键字连写等语法报错问题入参大模型原始SQL字符串返回清洗后可直接执行的标准英文SQLif 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*|\sFROM\b|\sWHERE\b|\sORDER\b|\sGROUP\b|\sLIMIT\b|\s*;),_clean_alias_space,sql,flagsre.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:# 字母下划线关键字粘连拆分补充第三个参数sqlpattern r([a-z_])( re.escape(kw) r)sql re.sub(pattern, r\1 \2, sql)# 中文文字关键字粘连拆分补充第三个参数sqlpattern_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,flagsre.IGNORECASE)# 最终再清理一遍多余空格sql re.sub(r\s, , sql).strip()return sqldef 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_inputuser_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:核心业务2SQL性能调优分析流程清洗SQL → 获取EXPLAIN执行计划 → AI分析给出优化方案入参用户输入待优化SQL返回执行状态、清洗后SQL、执行计划字段/内容、调优建议llm get_llm()db Mysql80Client()try:# 先标准化清洗SQLclean_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_inputclean_sql,explain_datastr(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(⚠️ 请输入有效需求)continueresult 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], headerskeys, tablefmtpretty))if result[summary]:print(f\n 业务总结\n{result[summary]})else:print(\n⚠️ 未查询到匹配数据)# 功能2SQL性能调优elif choice 2:sql_input input(\n请输入需要分析的 SQL 语句: ).strip()if not sql_input:print(⚠️ 请输入有效SQL)continueresult sql_tune_analyze(sql_input)if not result[success]:print(f\n❌ 分析失败{result[error]})continue# 打印执行计划表格和AI优化建议print(f\n 执行计划详情)print(tabulate(result[plan_rows], headerskeys, tablefmtpretty))print(f\n 调优建议\n{result[suggestion]})# 0 退出循环结束程序elif choice 0:print(程序已安全退出。)break# 无效数字输入提示else:print(无效输入请重试。)print(\n - * 40)# 程序入口直接运行main.py则启动终端菜单if __name__ __main__:main_cli()4.4创建 systemd 后台常驻服务(vim /etc/systemd/system/mysql-ai-web.service)[Unit]DescriptionINDODB AI Streamlit Web ToolAfternetwork.target mysqld.service[Service]TypesimpleUserrootWorkingDirectory/opt/mysql_ai_toolsExecStart/usr/local/python3.11/bin/python3 -m streamlit run web_main.py --server.address 0.0.0.0 --server.port 8501 --server.headless trueRestartalwaysRestartSec3StandardOutputjournalStandardErrorjournal[Install]WantedBymulti-user.target启动服务4.5.Windows网页访问测试4.5.1查看ip4.5.2网页测试(1)查询用户 1001 的所有订单展示商品名称、支付金额和下单时间(2)SELECT user_id AS 用户ID, SUM(pay_amount) AS 总消费金额, COUNT(id) AS 订单笔数 FROM order_info WHERE user_id IN (1001, 1002, 1003) GROUP BY user_idORDER BY 总消费金额 DESC