一、項目實施全流程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 數據庫開發工程師?！救蝿漳繕恕扛鶕脩糇匀徽Z言描述的業務需求生成可直接執行、無語法錯誤的MySQL查詢SQL?！颈斫Y構參考】{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執行計劃數據定位查詢性能問題并給出可直接落地的優化方案?！颈斫Y構參考】{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