一、項(xiàng)目實(shí)施全流程1.1openEuler 系統(tǒng)初始化配置1.1.1 系統(tǒng)安全與網(wǎng)絡(luò)優(yōu)化剛裝好 openEuler 系統(tǒng)防火墻、SELinux 會(huì)攔截端口訪問(wèn)時(shí)間同步錯(cuò)亂先做系統(tǒng)基礎(chǔ)優(yōu)化。1.1.2 源碼編譯 Python3.11.91.下載源碼包上傳至/usr/local/src解壓編譯系統(tǒng)自帶 Python3 版本過(guò)低選擇源碼編譯全新 Python1.1.3配置國(guó)內(nèi)阿里 pip 源國(guó)外官方 pip 源下載依賴速度極慢切換阿里鏡像站加速下載2.2 MySQL8.0.45 部署 測(cè)試訂單庫(kù)初始化2.1.1 創(chuàng)建訂單測(cè)試庫(kù)、表、20 條測(cè)試數(shù)據(jù)2.2 騰訊云 TokenHub 大模型平臺(tái)準(zhǔn)備1.進(jìn)入 TokenHub 平臺(tái) - API Key 管理新建 API 密鑰復(fù)制保存 sk 開(kāi)頭的密鑰2.模型廣場(chǎng)選擇deepseek-v4-pro二、項(xiàng)目 Python 源碼開(kāi)發(fā)逐文件帶功能 設(shè)計(jì)思路講解3.1 創(chuàng)建項(xiàng)目目錄3.2編寫環(huán)境變量腳本3.3 編寫數(shù)據(jù)庫(kù)連接腳本(vim /opt/mysql_ai_tools/mysql_client.py)# -*- coding: utf-8 -*-# 文件名mysql_client.py# 功能MySQL8.0數(shù)據(jù)庫(kù)統(tǒng)一封裝類# 作用封裝數(shù)據(jù)庫(kù)連接、普通查詢、EXPLAIN執(zhí)行計(jì)劃、SQL安全攔截統(tǒng)一拋出友好異常給上層業(yè)務(wù)調(diào)用import pymysqlimport osimport refrom dotenv import load_dotenv# 加載項(xiàng)目根目錄下.env文件的數(shù)據(jù)庫(kù)配置load_dotenv()class Mysql80Client:# 數(shù)據(jù)庫(kù)操作封裝類所有數(shù)據(jù)庫(kù)相關(guān)操作統(tǒng)一在此管理def __init__(self):# 初始化時(shí)讀取環(huán)境變量參數(shù)缺失則設(shè)置兜底默認(rèn)值防止程序直接崩潰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)# 數(shù)據(jù)庫(kù)連接對(duì)象初始為空self.conn None# 實(shí)例創(chuàng)建后自動(dòng)建立數(shù)據(jù)庫(kù)連接self.connect()def connect(self):創(chuàng)建數(shù)據(jù)庫(kù)連接捕獲連接異常并拋出可讀錯(cuò)誤信息try:self.conn pymysql.connect(hostself.host,portself.port,userself.user,passwordself.password,databaseself.database,charsetutf8mb4, # 支持中文、emoji完整字符集cursorclasspymysql.cursors.DictCursor # 查詢結(jié)果以字典返回方便按字段取值)except pymysql.MySQLError as e:# MySQL專屬連接錯(cuò)誤提示賬號(hào)、地址、密碼排查方向raise Exception(f數(shù)據(jù)庫(kù)連接失敗請(qǐng)檢查地址/賬號(hào)/密碼{e.args[1]})except Exception as e:# 其余未知連接異常統(tǒng)一捕獲raise Exception(f數(shù)據(jù)庫(kù)連接異常{str(e)})staticmethoddef _check_sql_safety(sql: str) - None:靜態(tài)私有安全校驗(yàn)方法核心防護(hù)攔截增刪改、建表刪表等危險(xiǎn)操作僅允許SELECT查詢防止AI生成危險(xiǎn)SQL篡改數(shù)據(jù)# 去除首尾空格并轉(zhuǎn)為大寫統(tǒng)一匹配規(guī)則sql_trim sql.strip().upper()# 危險(xiǎn)操作關(guān)鍵字黑名單danger_keywords [INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, TRUNCATE, REPLACE]for kw in danger_keywords:# 單詞邊界匹配避免字段名包含關(guān)鍵字時(shí)誤攔截if re.search(r\b re.escape(kw) r\b, sql_trim):raise Exception(f安全攔截禁止執(zhí)行 {kw} 類型語(yǔ)句僅支持 SELECT 查詢)def execute_query(self, sql: str):執(zhí)行普通SELECT查詢:param sql: 待執(zhí)行查詢語(yǔ)句:return: (字段名列表, 全部數(shù)據(jù)行字典列表)# 執(zhí)行SQL前先做安全校驗(yàn)攔截危險(xiǎn)語(yǔ)句self._check_sql_safety(sql)try:# with自動(dòng)管理游標(biāo)用完自動(dòng)釋放資源with self.conn.cursor() as cursor:cursor.execute(sql)# 提取查詢結(jié)果表頭字段columns [desc[0] for desc in cursor.description]# 讀取全部查詢數(shù)據(jù)rows cursor.fetchall()return columns, rowsexcept pymysql.MySQLError as e:# 捕獲SQL語(yǔ)法、表不存在等數(shù)據(jù)庫(kù)執(zhí)行錯(cuò)誤raise Exception(fSQL執(zhí)行失敗錯(cuò)誤碼 {e.args[0]}{e.args[1]})except Exception as e:# 通用查詢異常兜底raise Exception(f查詢異常{str(e)})def get_explain_plan(self, sql: str):獲取SQL執(zhí)行計(jì)劃EXPLAIN用于性能調(diào)優(yōu)分析:param sql: 待分析SELECT語(yǔ)句:return: (執(zhí)行計(jì)劃表頭, 執(zhí)行計(jì)劃詳情數(shù)據(jù))# 同樣先校驗(yàn)SQL安全性self._check_sql_safety(sql)# 拼接EXPLAIN關(guān)鍵字生成分析語(yǔ)句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獲取執(zhí)行計(jì)劃失敗{e.args[1]})except Exception as e:raise Exception(f執(zhí)行計(jì)劃異常{str(e)})def close(self):安全關(guān)閉數(shù)據(jù)庫(kù)連接釋放資源避免長(zhǎng)時(shí)間占用連接池# 判斷連接存在且未關(guān)閉才執(zhí)行關(guān)閉操作if self.conn and not self.conn._closed:self.conn.close()3.4. 編寫sql轉(zhuǎn)換腳本(vim /opt/mysql_ai_tools/prompts.py)# -*- coding: utf-8 -*-# 文件名prompts.py# 功能統(tǒng)一管理項(xiàng)目全部大模型提示詞模板附帶SQL提取工具靜態(tài)方法# 作用把AI提示詞和業(yè)務(wù)代碼解耦統(tǒng)一約束模型輸出格式降低SQL解析報(bào)錯(cuò)概率import reclass UnifiedPrompt:提示詞統(tǒng)一管理類優(yōu)勢(shì)所有SQL生成、性能分析提示詞集中存放表結(jié)構(gòu)僅維護(hù)一處修改不用多處同步通過(guò)嚴(yán)格規(guī)則約束大模型輸出減少格式錯(cuò)亂、編造字段、危險(xiǎn)SQL等幻覺(jué)問(wèn)題# 全局共用數(shù)據(jù)表結(jié)構(gòu) # 只在此維護(hù)訂單表字段下方兩套提示詞會(huì)自動(dòng)引用改表結(jié)構(gòu)只需改這里一處TABLE_SCHEMA 表名: order_info (訂單信息表)字段說(shuō)明:- id: 訂單ID (主鍵INT類型)- user_id: 用戶ID (INT類型)- order_name: 商品名稱 (VARCHAR類型)- pay_amount: 支付金額 (DECIMAL類型)- create_time: 下單時(shí)間 (DATETIME類型)# 模板1自然語(yǔ)言轉(zhuǎn)SQL專用提示詞 NL_TO_SQL_PROMPT f你是嚴(yán)謹(jǐn)?shù)?MySQL 8.0 數(shù)據(jù)庫(kù)開(kāi)發(fā)工程師?！救蝿?wù)目標(biāo)】根據(jù)用戶自然語(yǔ)言描述的業(yè)務(wù)需求生成可直接執(zhí)行、無(wú)語(yǔ)法錯(cuò)誤的MySQL查詢SQL。【表結(jié)構(gòu)參考】{TABLE_SCHEMA}【強(qiáng)制輸出規(guī)則】1. 只能生成 SELECT 查詢語(yǔ)句絕對(duì)不允許生成 INSERT/UPDATE/DELETE/DROP 等修改、刪除數(shù)據(jù)的語(yǔ)句。2. 只能使用上面列出的5個(gè)字段禁止自己編造不存在的字段名。3. 查詢字段可使用中文別名格式固定為字段 AS 別名。4 SQL語(yǔ)法遵循MySQL8.0標(biāo)準(zhǔn)所有關(guān)鍵字統(tǒng)一大寫方便程序解析。5. 最終SQL必須包裹在 sql Markdown代碼塊內(nèi)方便代碼提取。6. 禁止輸出任何解釋、說(shuō)明文字只返回純SQL代碼塊減少解析干擾。7. 中文別名內(nèi)部不能帶空格例訂單ID正確、訂單 ID錯(cuò)誤避免數(shù)據(jù)庫(kù)語(yǔ)法報(bào)錯(cuò)。【用戶需求】{{user_input}}# 模板2SQL性能調(diào)優(yōu)分析專用提示詞 SQL_TUNE_PROMPT f你是資深 MySQL DBA 性能優(yōu)化專家。【任務(wù)目標(biāo)】根據(jù)原始SQL EXPLAIN執(zhí)行計(jì)劃數(shù)據(jù)定位查詢性能問(wèn)題并給出可直接落地的優(yōu)化方案?！颈斫Y(jié)構(gòu)參考】{TABLE_SCHEMA}【待分析SQL】{{sql_input}}【執(zhí)行計(jì)劃數(shù)據(jù)】{{explain_data}}【輸出要求】1. 先點(diǎn)明核心性能問(wèn)題全表掃描、無(wú)索引、索引失效、掃描行數(shù)過(guò)多等。2. 給出完整建索引SQL語(yǔ)句可直接復(fù)制執(zhí)行。3. 若原SQL寫法存在缺陷提供改寫后的完整優(yōu)化SQL。4. 內(nèi)容簡(jiǎn)潔、分點(diǎn)羅列不輸出多余廢話便于用戶快速閱讀。staticmethoddef extract_sql(response_text: str) - str:靜態(tài)工具方法從大模型返回的完整文本里剝離出純凈SQL語(yǔ)句三層匹配優(yōu)先級(jí)兼容不同大模型的輸出格式提升提取成功率:param response_text: 大模型原始完整返回內(nèi)容:return: 清洗后的純SQL字符串提取失敗返回空字符串# 空文本直接返回if not response_text:return # 優(yōu)先級(jí)1匹配最標(biāo)準(zhǔn)markdown sql代碼塊項(xiàng)目提示詞強(qiáng)制要求的格式match re.search(rsql\s*(.*?)\s*, response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 優(yōu)先級(jí)2兼容自定義sql標(biāo)簽格式備用兼容方案match re.search(rsql\s*(.*?)\s*/sql, response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 優(yōu)先級(jí)3兜底匹配直接抓取以SELECT開(kāi)頭、分號(hào)結(jié)尾的SQL片段match re.search(r(SELECT\s.*?;), response_text, re.DOTALL | re.IGNORECASE)if match:return match.group(1).strip()# 三層規(guī)則全部匹配不到說(shuō)明無(wú)有效SQL返回空return # 全局單例實(shí)例外部文件導(dǎo)入后直接調(diào)用 prompt_helper.方法名無(wú)需重復(fù)實(shí)例化prompt_helper UnifiedPrompt()3.5. 編寫程序入口腳本(vim /opt/mysql_ai_tools/main.py)# -*- coding: utf-8 -*-# 文件名main.py# 功能項(xiàng)目核心業(yè)務(wù)邏輯 終端交互式菜單入口# 作用統(tǒng)一封裝大模型調(diào)用、SQL清洗、數(shù)據(jù)庫(kù)交互兩大核心業(yè)務(wù)命令行/網(wǎng)頁(yè)共用底層函數(shù)import osimport reimport loggingfrom dotenv import load_dotenv# 兼容OpenAI標(biāo)準(zhǔn)大模型接口適配騰訊云TokenHubfrom langchain_openai import ChatOpenAI# 導(dǎo)入數(shù)據(jù)庫(kù)操作封裝類from mysql_client import Mysql80Client# 表格格式化打印工具美化終端輸出查詢結(jié)果from tabulate import tabulate# 導(dǎo)入提示詞管理類與全局實(shí)例from prompts import UnifiedPrompt, prompt_helper# 加載.env文件里所有數(shù)據(jù)庫(kù)、大模型配置load_dotenv()# 全局日志配置替代print記錄運(yùn)行時(shí)間、日志級(jí)別、報(bào)錯(cuò)信息方便排障logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s)logger logging.getLogger(__name__)def check_config() - None:程序啟動(dòng)前置配置校驗(yàn)函數(shù)作用提前檢測(cè).env必填參數(shù)是否存在避免運(yùn)行中途缺參數(shù)崩潰# 大模型必填參數(shù)列表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配置缺失請(qǐng)?jiān)?.env 文件中填寫 {, .join(missing)})# 數(shù)據(jù)庫(kù)必填參數(shù)列表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數(shù)據(jù)庫(kù)配置缺失請(qǐng)檢查 {, .join(missing_db)})def get_llm() - ChatOpenAI:初始化大模型客戶端適配騰訊云TokenHub等全部兼容OpenAI接口規(guī)范的MaaS平臺(tái)返回可直接調(diào)用的大模型實(shí)例# 從環(huán)境變量讀取大模型連接信息api_key os.getenv(LLM_API_KEY)base_url os.getenv(LLM_BASE_URL)model_name os.getenv(LLM_MODEL_NAME)# 溫度不存在則默認(rèn)0.1數(shù)值越低輸出越嚴(yán)謹(jǐn)穩(wěn)定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標(biāo)準(zhǔn)化清洗工具函數(shù)兜底修復(fù)各大模型輸出格式解決中文空格別名、中文標(biāo)點(diǎn)、特殊空白、關(guān)鍵字連寫等語(yǔ)法報(bào)錯(cuò)問(wèn)題入?yún)⒋竽Ｐ驮糞QL字符串返回清洗后可直接執(zhí)行的標(biāo)準(zhǔn)英文SQLif not sql:return # 1. 統(tǒng)一替換各類中文全角空格、換行、制表符為普通半角空格special_spaces [\xa0, \u200b, \u200c, \u200d, \u200e, \u200f,\u3000, \t, \n, \r]for sp in special_spaces:sql sql.replace(sp, )# 2. 刪除不可見(jiàn)控制字符防止解析異常sql re.sub(r[\x00-\x1f\x7f], , sql)# 3. 中文標(biāo)點(diǎn)批量替換為英文標(biāo)點(diǎn)解決Qwen等模型輸出中文逗號(hào)報(bào)錯(cuò)sql sql.replace(, ,).replace(, ;).replace(, ().replace(, ))# 4. 多個(gè)連續(xù)空格合并為單個(gè)去除首尾多余空格sql re.sub(r\s, , sql).strip()# 5. 精準(zhǔn)處理AS別名內(nèi)部空格只刪別名里空格保留AS與別名之間分隔空格def _clean_alias_space(match):prefix match.group(1) # 捕獲AS關(guān)鍵字alias match.group(2) # 捕獲后面全部別名文本alias_clean re.sub(r\s, , alias)return f{prefix} {alias_clean}# 匹配AS后別名截止逗號(hào)、FROM、WHERE等關(guān)鍵字前停止匹配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. 自動(dòng)給連寫的關(guān)鍵字補(bǔ)空格字段/中文關(guān)鍵字粘連自動(dòng)拆分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:# 字母下劃線關(guān)鍵字粘連拆分補(bǔ)充第三個(gè)參數(shù)sqlpattern r([a-z_])( re.escape(kw) r)sql re.sub(pattern, r\1 \2, sql)# 中文文字關(guān)鍵字粘連拆分補(bǔ)充第三個(gè)參數(shù)sqlpattern_cn r([\u4e00-\u9fa5])( re.escape(kw) r)sql re.sub(pattern_cn, r\1 \2, sql)# 7. 統(tǒng)一所有SQL關(guān)鍵字大寫格式標(biāo)準(zhǔn)化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:核心業(yè)務(wù)1自然語(yǔ)言轉(zhuǎn)SQL、執(zhí)行查詢、AI生成業(yè)務(wù)總結(jié)對(duì)外統(tǒng)一標(biāo)準(zhǔn)返回字典終端/網(wǎng)頁(yè)程序均可直接調(diào)用無(wú)重復(fù)代碼入?yún)⒂脩糇匀徽Z(yǔ)言查詢需求返回包含執(zhí)行狀態(tài)、SQL、字段、數(shù)據(jù)、AI總結(jié)、模型原始輸出# 初始化大模型、數(shù)據(jù)庫(kù)客戶端llm get_llm()db Mysql80Client()try:logger.info(正在生成SQL語(yǔ)句...)# 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請(qǐng)重新描述需求)# 3. 清洗SQL修復(fù)各類格式問(wèn)題clean_sql clean_sql_spacing(extracted_sql)logger.info(f生成SQL{clean_sql})# 4. 數(shù)據(jù)庫(kù)執(zhí)行查詢拿到表頭與數(shù)據(jù)columns, rows db.execute_query(clean_sql)# 5. 如果有數(shù)據(jù)調(diào)用大模型生成業(yè)務(wù)解讀總結(jié)summary if rows:logger.info(正在生成數(shù)據(jù)總結(jié)...)summary_prompt f以下是真實(shí)的SQL查詢結(jié)果請(qǐng)作為電商數(shù)據(jù)分析師給出簡(jiǎn)練的業(yè)務(wù)總結(jié)。SQL語(yǔ)句{clean_sql}查詢數(shù)據(jù){str(rows)}重點(diǎn)說(shuō)明數(shù)據(jù)反映的業(yè)務(wù)含義如有異常值請(qǐng)指出。summary_resp llm.invoke(summary_prompt)summary summary_resp.content.strip()# 成功結(jié)果返回return {success: True,sql: clean_sql,columns: columns,rows: rows,summary: summary,raw_llm: raw_content}except Exception as e:# 捕獲全流程所有異常記錄日志并返回錯(cuò)誤信息logger.error(f查詢處理失敗{str(e)})return {success: False,error: str(e),raw_llm: raw_content if raw_content in dir() else }finally:# 無(wú)論成功失敗都關(guān)閉數(shù)據(jù)庫(kù)連接釋放資源db.close()def sql_tune_analyze(raw_sql: str) - dict:核心業(yè)務(wù)2SQL性能調(diào)優(yōu)分析流程清洗SQL → 獲取EXPLAIN執(zhí)行計(jì)劃 → AI分析給出優(yōu)化方案入?yún)⒂脩糨斎氪齼?yōu)化SQL返回執(zhí)行狀態(tài)、清洗后SQL、執(zhí)行計(jì)劃字段/內(nèi)容、調(diào)優(yōu)建議llm get_llm()db Mysql80Client()try:# 先標(biāo)準(zhǔn)化清洗SQLclean_sql clean_sql_spacing(raw_sql)logger.info(正在獲取執(zhí)行計(jì)劃...)# 調(diào)用數(shù)據(jù)庫(kù)封裝方法獲取EXPLAIN執(zhí)行計(jì)劃columns, plan_rows db.get_explain_plan(clean_sql)# 填充調(diào)優(yōu)提示詞傳入SQL和執(zhí)行計(jì)劃讓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調(diào)優(yōu)分析失敗{str(e)})return {success: False,error: str(e)}finally:# 操作結(jié)束關(guān)閉數(shù)據(jù)庫(kù)連接db.close()def main_cli():終端交互入口主函數(shù)提供循環(huán)菜單支持用戶選擇查詢/調(diào)優(yōu)/退出純終端操作# 程序啟動(dòng)先校驗(yàn)全部配置失敗直接退出菜單try:check_config()except ValueError as e:print(f? {e})return# 循環(huán)交互不退出可持續(xù)多次使用while True:print(\n InnoAI SQL 助手 )print(1. 自然語(yǔ)言生成SQL自動(dòng)查詢并AI總結(jié)數(shù)據(jù))print(2. 輸入SQL語(yǔ)句AI分析執(zhí)行計(jì)劃并給出調(diào)優(yōu)方案)print(0. 退出程序)choice input(請(qǐng)輸入功能序號(hào): ).strip()# 功能1自然語(yǔ)言查數(shù)據(jù)if choice 1:query input(請(qǐng)輸入你的數(shù)據(jù)查詢需求: ).strip()if not query:print(?? 請(qǐng)輸入有效需求)continueresult nl2sql_query(query)# 處理失敗場(chǎng)景打印錯(cuò)誤與模型原始輸出if not result[success]:print(f\n? 處理失敗{result[error]})if result.get(raw_llm):print(f大模型原始回復(fù)\n{result[raw_llm]})continue# 成功打印SQL、格式化表格展示數(shù)據(jù)、輸出業(yè)務(wù)總結(jié)print(f\n? 生成SQL)print(result[sql])if result[rows]:print(f\n 查詢結(jié)果共 {len(result[rows])} 條)print(tabulate(result[rows], headerskeys, tablefmtpretty))if result[summary]:print(f\n 業(yè)務(wù)總結(jié)\n{result[summary]})else:print(\n?? 未查詢到匹配數(shù)據(jù))# 功能2SQL性能調(diào)優(yōu)elif choice 2:sql_input input(\n請(qǐng)輸入需要分析的 SQL 語(yǔ)句: ).strip()if not sql_input:print(?? 請(qǐng)輸入有效SQL)continueresult sql_tune_analyze(sql_input)if not result[success]:print(f\n? 分析失敗{result[error]})continue# 打印執(zhí)行計(jì)劃表格和AI優(yōu)化建議print(f\n 執(zhí)行計(jì)劃詳情)print(tabulate(result[plan_rows], headerskeys, tablefmtpretty))print(f\n 調(diào)優(yōu)建議\n{result[suggestion]})# 0 退出循環(huán)結(jié)束程序elif choice 0:print(程序已安全退出。)break# 無(wú)效數(shù)字輸入提示else:print(無(wú)效輸入請(qǐng)重試。)print(\n - * 40)# 程序入口直接運(yùn)行main.py則啟動(dòng)終端菜單if __name__ __main__:main_cli()三、測(cè)試1.查詢用戶 1001 的所有訂單展示商品名稱、支付金額和下單時(shí)間2.SELECT user_id AS 用戶ID, SUM(pay_amount) AS 總消費(fèi)金額, COUNT(id) AS 訂單筆數(shù) FROM order_info WHERE user_id IN (1001, 1002, 1003) GROUP BY user_idORDER BY 總消費(fèi)金額 DESC四、web可視化版本4.1 web_main.py Streamlit 可視化網(wǎng)頁(yè)功能說(shuō)明基于 Streamlit 實(shí)現(xiàn)可視化網(wǎng)頁(yè)界面復(fù)用 main.py 封裝好的業(yè)務(wù)函數(shù)提供瀏覽器遠(yuǎn)程操作入口為什么這么設(shè)計(jì)1.純 Python 開(kāi)發(fā)網(wǎng)頁(yè)零基礎(chǔ)快速實(shí)現(xiàn)可視化頁(yè)面作為項(xiàng)目拓展加分功能2.頁(yè)面做輸入前置校驗(yàn)區(qū)分自然語(yǔ)言輸入框和 SQL 輸入框防止用戶操作混淆3.美化頁(yè)面樣式表格、代碼塊、按鈕優(yōu)化演示項(xiàng)目時(shí)觀感更好4.支持展開(kāi)面板查看大模型原始返回方便調(diào)試排錯(cuò)。4.2安裝 Python 依賴4.3創(chuàng)建 Streamlit 可視化入口 web_main.pyvim /opt/mysql_ai_tools/web_main.py# -*- coding: utf-8 -*-# 文件名main.py# 功能項(xiàng)目核心業(yè)務(wù)邏輯 終端交互式菜單入口# 作用統(tǒng)一封裝大模型調(diào)用、SQL清洗、數(shù)據(jù)庫(kù)交互兩大核心業(yè)務(wù)命令行/網(wǎng)頁(yè)共用底層函數(shù)import osimport reimport loggingfrom dotenv import load_dotenv# 兼容OpenAI標(biāo)準(zhǔn)大模型接口適配騰訊云TokenHubfrom langchain_openai import ChatOpenAI# 導(dǎo)入數(shù)據(jù)庫(kù)操作封裝類from mysql_client import Mysql80Client# 表格格式化打印工具美化終端輸出查詢結(jié)果from tabulate import tabulate# 導(dǎo)入提示詞管理類與全局實(shí)例from prompts import UnifiedPrompt, prompt_helper# 加載.env文件里所有數(shù)據(jù)庫(kù)、大模型配置load_dotenv()# 全局日志配置替代print記錄運(yùn)行時(shí)間、日志級(jí)別、報(bào)錯(cuò)信息方便排障logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s)logger logging.getLogger(__name__)def check_config() - None:程序啟動(dòng)前置配置校驗(yàn)函數(shù)作用提前檢測(cè).env必填參數(shù)是否存在避免運(yùn)行中途缺參數(shù)崩潰# 大模型必填參數(shù)列表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配置缺失請(qǐng)?jiān)?.env 文件中填寫 {, .join(missing)})# 數(shù)據(jù)庫(kù)必填參數(shù)列表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數(shù)據(jù)庫(kù)配置缺失請(qǐng)檢查 {, .join(missing_db)})def get_llm() - ChatOpenAI:初始化大模型客戶端適配騰訊云TokenHub等全部兼容OpenAI接口規(guī)范的MaaS平臺(tái)返回可直接調(diào)用的大模型實(shí)例# 從環(huán)境變量讀取大模型連接信息api_key os.getenv(LLM_API_KEY)base_url os.getenv(LLM_BASE_URL)model_name os.getenv(LLM_MODEL_NAME)# 溫度不存在則默認(rèn)0.1數(shù)值越低輸出越嚴(yán)謹(jǐn)穩(wěn)定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標(biāo)準(zhǔn)化清洗工具函數(shù)兜底修復(fù)各大模型輸出格式解決中文空格別名、中文標(biāo)點(diǎn)、特殊空白、關(guān)鍵字連寫等語(yǔ)法報(bào)錯(cuò)問(wèn)題入?yún)⒋竽Ｐ驮糞QL字符串返回清洗后可直接執(zhí)行的標(biāo)準(zhǔn)英文SQLif not sql:return # 1. 統(tǒng)一替換各類中文全角空格、換行、制表符為普通半角空格special_spaces [\xa0, \u200b, \u200c, \u200d, \u200e, \u200f,\u3000, \t, \n, \r]for sp in special_spaces:sql sql.replace(sp, )# 2. 刪除不可見(jiàn)控制字符防止解析異常sql re.sub(r[\x00-\x1f\x7f], , sql)# 3. 中文標(biāo)點(diǎn)批量替換為英文標(biāo)點(diǎn)解決Qwen等模型輸出中文逗號(hào)報(bào)錯(cuò)sql sql.replace(, ,).replace(, ;).replace(, ().replace(, ))# 4. 多個(gè)連續(xù)空格合并為單個(gè)去除首尾多余空格sql re.sub(r\s, , sql).strip()# 5. 精準(zhǔn)處理AS別名內(nèi)部空格只刪別名里空格保留AS與別名之間分隔空格def _clean_alias_space(match):prefix match.group(1) # 捕獲AS關(guān)鍵字alias match.group(2) # 捕獲后面全部別名文本alias_clean re.sub(r\s, , alias)return f{prefix} {alias_clean}# 匹配AS后別名截止逗號(hào)、FROM、WHERE等關(guān)鍵字前停止匹配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. 自動(dòng)給連寫的關(guān)鍵字補(bǔ)空格字段/中文關(guān)鍵字粘連自動(dòng)拆分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:# 字母下劃線關(guān)鍵字粘連拆分補(bǔ)充第三個(gè)參數(shù)sqlpattern r([a-z_])( re.escape(kw) r)sql re.sub(pattern, r\1 \2, sql)# 中文文字關(guān)鍵字粘連拆分補(bǔ)充第三個(gè)參數(shù)sqlpattern_cn r([\u4e00-\u9fa5])( re.escape(kw) r)sql re.sub(pattern_cn, r\1 \2, sql)# 7. 統(tǒng)一所有SQL關(guān)鍵字大寫格式標(biāo)準(zhǔn)化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:核心業(yè)務(wù)1自然語(yǔ)言轉(zhuǎn)SQL、執(zhí)行查詢、AI生成業(yè)務(wù)總結(jié)對(duì)外統(tǒng)一標(biāo)準(zhǔn)返回字典終端/網(wǎng)頁(yè)程序均可直接調(diào)用無(wú)重復(fù)代碼入?yún)⒂脩糇匀徽Z(yǔ)言查詢需求返回包含執(zhí)行狀態(tài)、SQL、字段、數(shù)據(jù)、AI總結(jié)、模型原始輸出# 初始化大模型、數(shù)據(jù)庫(kù)客戶端llm get_llm()db Mysql80Client()try:logger.info(正在生成SQL語(yǔ)句...)# 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請(qǐng)重新描述需求)# 3. 清洗SQL修復(fù)各類格式問(wèn)題clean_sql clean_sql_spacing(extracted_sql)logger.info(f生成SQL{clean_sql})# 4. 數(shù)據(jù)庫(kù)執(zhí)行查詢拿到表頭與數(shù)據(jù)columns, rows db.execute_query(clean_sql)# 5. 如果有數(shù)據(jù)調(diào)用大模型生成業(yè)務(wù)解讀總結(jié)summary if rows:logger.info(正在生成數(shù)據(jù)總結(jié)...)summary_prompt f以下是真實(shí)的SQL查詢結(jié)果請(qǐng)作為電商數(shù)據(jù)分析師給出簡(jiǎn)練的業(yè)務(wù)總結(jié)。SQL語(yǔ)句{clean_sql}查詢數(shù)據(jù){str(rows)}重點(diǎn)說(shuō)明數(shù)據(jù)反映的業(yè)務(wù)含義如有異常值請(qǐng)指出。summary_resp llm.invoke(summary_prompt)summary summary_resp.content.strip()# 成功結(jié)果返回return {success: True,sql: clean_sql,columns: columns,rows: rows,summary: summary,raw_llm: raw_content}except Exception as e:# 捕獲全流程所有異常記錄日志并返回錯(cuò)誤信息logger.error(f查詢處理失敗{str(e)})return {success: False,error: str(e),raw_llm: raw_content if raw_content in dir() else }finally:# 無(wú)論成功失敗都關(guān)閉數(shù)據(jù)庫(kù)連接釋放資源db.close()def sql_tune_analyze(raw_sql: str) - dict:核心業(yè)務(wù)2SQL性能調(diào)優(yōu)分析流程清洗SQL → 獲取EXPLAIN執(zhí)行計(jì)劃 → AI分析給出優(yōu)化方案入?yún)⒂脩糨斎氪齼?yōu)化SQL返回執(zhí)行狀態(tài)、清洗后SQL、執(zhí)行計(jì)劃字段/內(nèi)容、調(diào)優(yōu)建議llm get_llm()db Mysql80Client()try:# 先標(biāo)準(zhǔn)化清洗SQLclean_sql clean_sql_spacing(raw_sql)logger.info(正在獲取執(zhí)行計(jì)劃...)# 調(diào)用數(shù)據(jù)庫(kù)封裝方法獲取EXPLAIN執(zhí)行計(jì)劃columns, plan_rows db.get_explain_plan(clean_sql)# 填充調(diào)優(yōu)提示詞傳入SQL和執(zhí)行計(jì)劃讓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調(diào)優(yōu)分析失敗{str(e)})return {success: False,error: str(e)}finally:# 操作結(jié)束關(guān)閉數(shù)據(jù)庫(kù)連接db.close()def main_cli():終端交互入口主函數(shù)提供循環(huán)菜單支持用戶選擇查詢/調(diào)優(yōu)/退出純終端操作# 程序啟動(dòng)先校驗(yàn)全部配置失敗直接退出菜單try:check_config()except ValueError as e:print(f? {e})return# 循環(huán)交互不退出可持續(xù)多次使用while True:print(\n InnoAI SQL 助手 )print(1. 自然語(yǔ)言生成SQL自動(dòng)查詢并AI總結(jié)數(shù)據(jù))print(2. 輸入SQL語(yǔ)句AI分析執(zhí)行計(jì)劃并給出調(diào)優(yōu)方案)print(0. 退出程序)choice input(請(qǐng)輸入功能序號(hào): ).strip()# 功能1自然語(yǔ)言查數(shù)據(jù)if choice 1:query input(請(qǐng)輸入你的數(shù)據(jù)查詢需求: ).strip()if not query:print(?? 請(qǐng)輸入有效需求)continueresult nl2sql_query(query)# 處理失敗場(chǎng)景打印錯(cuò)誤與模型原始輸出if not result[success]:print(f\n? 處理失敗{result[error]})if result.get(raw_llm):print(f大模型原始回復(fù)\n{result[raw_llm]})continue# 成功打印SQL、格式化表格展示數(shù)據(jù)、輸出業(yè)務(wù)總結(jié)print(f\n? 生成SQL)print(result[sql])if result[rows]:print(f\n 查詢結(jié)果共 {len(result[rows])} 條)print(tabulate(result[rows], headerskeys, tablefmtpretty))if result[summary]:print(f\n 業(yè)務(wù)總結(jié)\n{result[summary]})else:print(\n?? 未查詢到匹配數(shù)據(jù))# 功能2SQL性能調(diào)優(yōu)elif choice 2:sql_input input(\n請(qǐng)輸入需要分析的 SQL 語(yǔ)句: ).strip()if not sql_input:print(?? 請(qǐng)輸入有效SQL)continueresult sql_tune_analyze(sql_input)if not result[success]:print(f\n? 分析失敗{result[error]})continue# 打印執(zhí)行計(jì)劃表格和AI優(yōu)化建議print(f\n 執(zhí)行計(jì)劃詳情)print(tabulate(result[plan_rows], headerskeys, tablefmtpretty))print(f\n 調(diào)優(yōu)建議\n{result[suggestion]})# 0 退出循環(huán)結(jié)束程序elif choice 0:print(程序已安全退出。)break# 無(wú)效數(shù)字輸入提示else:print(無(wú)效輸入請(qǐng)重試。)print(\n - * 40)# 程序入口直接運(yùn)行main.py則啟動(dòng)終端菜單if __name__ __main__:main_cli()4.4創(chuàng)建 systemd 后臺(tái)常駐服務(wù)(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啟動(dòng)服務(wù)4.5.Windows網(wǎng)頁(yè)訪問(wèn)測(cè)試4.5.1查看ip4.5.2網(wǎng)頁(yè)測(cè)試(1)查詢用戶 1001 的所有訂單展示商品名稱、支付金額和下單時(shí)間(2)SELECT user_id AS 用戶ID, SUM(pay_amount) AS 總消費(fèi)金額, COUNT(id) AS 訂單筆數(shù) FROM order_info WHERE user_id IN (1001, 1002, 1003) GROUP BY user_idORDER BY 總消費(fèi)金額 DESC