一、环境搭建1.利用Xftp安装python3.11.9,将python.tgz拖入/usr/local/src目录2.解压python包tar -zxvf Python-3.11.9.tgz3.安装依赖环境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-server4.编译配置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验证安装成功[global] index-url http://mirrors.aliyun.com/pypi/simple/ [install] trusted-hostmirrors.aliyun.com校验SSL模块python3 -c import ssl; print(ssl.OPENSSL_VERSION)5.配置国内 pip 源及安装项目依赖配置pip仓库为阿里镜像站提高下载速度vim ~/.pip/pip.conf[global] index-url http://mirrors.aliyun.com/pypi/simple/ [install] trusted-hostmirrors.aliyun.com安装python及ai所需依赖pip3 install --upgrade pip pip3 install pymysql python-dotenv tabulate langchain langchain-openai6.部署数据库创建库 / 表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 下单时间 ) ENGINEInnoDB COMMENT电商订单业务表二、获取API Key访问腾讯云官网新建 API 密钥三、编写脚本1. 新建目录及脚本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 12.编写脚本.env# MySQL 数据库连接配置 # 数据库服务IP本地运行填127.0.0.1远程服务器替换对应公网/内网地址 MYSQL_HOST127.0.0.1 # MySQL默认通信端口未手动修改固定3306 MYSQL_PORT3306 # 数据库登录用户名测试环境默认root MYSQL_USERroot # 数据库登录密码根据自己本地MySQL密码修改 MYSQL_PASSWORD123456 # 项目专用数据库名SQL查询仅操作该库下order_info表 MYSQL_DBtestdb # 腾讯云TokenHub大模型配置 # TokenHub平台分配的API密钥用于鉴权调用大模型切勿泄露 LLM_API_KEYsk-jUJebmj56mEQhhBep04A3AIaVlVV2NA97F45iimZ*** # TokenHub统一接口地址固定/v1后缀兼容OpenAI标准SDK LLM_BASE_URLhttps://tokenhub.tencentmaas.com/v1 # 当前调用模型标识切换DeepSeek v4 Pro只需改为deepseek-v4-pro LLM_MODEL_NAMEdeepseek-v4-pro # 模型温度参数0代表输出最严谨稳定适合生成标准SQL不产生随机偏差 LLM_TEMPERATURE0 [rootserver ~]# chmod 600 /opt/mysql_ai_tools/.envmysql_client.py#-*- coding: utf-8 -* # 文件名mysql_client.py # 功能MySQL8.0数据库统一封装类 # 作用封装数据库连接、普通查询、EXPLAIN执行计划、SQL安全拦截统一抛出友好异常给上层业务调用 import pymysql import os import re from 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)}) 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: (字段名列表, 全部数据行字典列表) # 执行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, rows except pymysql.MySQLError as e: 5.3.4. 编写sql转换脚本 # 捕获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, 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()prompts.py# -*- coding: utf-8 -* # 文件名prompts.py # 功能统一管理项目全部大模型提示词模板附带SQL提取工具静态方法 # 作用把AI提示词和业务代码解耦统一约束模型输出格式降低SQL解析报错概率 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类型) # 模板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. 内容简洁、分点罗列不输出多余废话便于用户快速阅读。 staticmethod def 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()main.py# -*- coding: utf-8 -* # 文件名main.py # 功能项目核心业务逻辑 终端交互式菜单入口 # 作用统一封装大模型调用、SQL清洗、数据库交互两大核心业务命令行/网页共用底层函数 import os import re import logging from dotenv import load_dotenv # 兼容OpenAI标准大模型接口适配腾讯云TokenHub from langchain_openai import ChatOpenAI # 导入数据库操作封装类 from mysql_client import Mysql80Client # 表格格式化打印工具美化终端输出查询结果 from tabulate import tabulate # 导入提示词管理类与全局实例 from prompts import UnifiedPrompt, prompt_helper # 全局日志配置替代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字符串 返回清洗后可直接执行的标准英文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*|\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: # 字母下划线关键字粘连拆分补充第三个参数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, flagsre.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_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: # 先标准化清洗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_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( ⚠ 请输入有效需求) 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], 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) 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], 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()