SQL 語法糾錯循環(huán)基于數(shù)據(jù)庫執(zhí)行報錯的 Agent 動態(tài)自愈在 Text2SQL 的真實生產(chǎn)落地中即便前置完成了精準的 Schema 召回與 Few-Shot 注入大模型一次性寫出 100% 完美無瑕 SQL 的概率依然很難突破 80%。大模型生成的 SQL 常常會遭遇各種數(shù)據(jù)庫底層的剛性報錯字段或表名拼寫錯誤如誤寫為user_name而實際列名為username聚合函數(shù)與 GROUP BY 缺失如在SELECT dept_id, count(*)中遺漏了GROUP BY dept_id數(shù)據(jù)類型隱式轉換失敗如將字符串與時間戳直接做大于等于比較方言特有語法不兼容在 ClickHouse 中誤用了 MySQL 的專有函數(shù)。如果系統(tǒng)在遇到數(shù)據(jù)庫報錯時直接把錯誤信息拋給最終用戶整個系統(tǒng)的可用性將大打折扣。構建一個**“生成 - 預執(zhí)行 - 捕獲報錯 - 注入反思 - 動態(tài)自糾錯Self-Correction Loop”**的閉環(huán)機制是保障 Text2SQL 端到端執(zhí)行成功率突破 95% 的核心技術底牌。一、動態(tài)自愈閉環(huán)的執(zhí)行流架構[ 用戶自然語言 Query ] │ ▼ ┌─────────────────────────────────┐ │ 1. 初次 SQL 生成 (Initial Plan) │ └────────────────┬────────────────┘ │ ▼ ┌─────────────────────────────────┐ │ 2. 只讀沙箱執(zhí)行 (Dry-Run Guard) │ ──(帶有 LIMIT 1 與只讀事務保護) └────────────────┬────────────────┘ │ ┌────────┴────────┐ ▼ ▼ [ 執(zhí)行成功 (Success) ] [ 捕獲 DB 異常 (Error Traced) ] 格式化輸出業(yè)務結論 提取精準錯誤碼與詳細堆棧 │ ▼ ┌─────────────────────────────────┐ │ 3. 錯誤提示詞組裝與上下文增強 │ │ (結合原始 SQL 報錯原因 Schema)│ └────────────────┬────────────────┘ │ ▼ ┌─────────────────────────────────┐ │ 4. 大模型自糾錯 (Re-generate) │ ──(最大重試 2 輪) └─────────────────────────────────┘二、生產(chǎn)級自糾錯 Agent 的 Python 實現(xiàn)import sqlite3 from typing import Dict, Any, Optional from pydantic import BaseModel class SQLCorrectionState(BaseModel): natural_query: str schema_info: str current_sql: str retry_count: int 0 max_retries: int 2 is_success: bool False result_data: Optional[list] None last_error: Optional[str] None class Text2SQLSelfHealer: def __init__(self, llm_client, db_connection): self.llm llm_client self.db db_connection def execute_and_heal(self, query: str, schema_info: str) - SQLCorrectionState: # 1. 初次生成 initial_sql self._generate_sql(query, schema_info) state SQLCorrectionState( natural_queryquery, schema_infoschema_info, current_sqlinitial_sql ) while state.retry_count state.max_retries: # 2. 嘗試在數(shù)據(jù)庫中執(zhí)行驗證 (帶有只讀攔截) success, result_or_err self._try_execute_sql(state.current_sql) if success: state.is_success True state.result_data result_or_err return state # 3. 記錄報錯并觸發(fā)自愈推理 state.last_error result_or_err state.retry_count 1 if state.retry_count state.max_retries: break # 超過最大自愈次數(shù)終止 # 4. 生成修正后的新 SQL corrected_sql self._heal_sql(state) state.current_sql corrected_sql return state def _try_execute_sql(self, sql: str) - tuple[bool, Any]: 安全預執(zhí)行限制只讀與返回條數(shù) clean_sql sql.strip().rstrip(;) if not clean_sql.upper().startswith(SELECT): return False, 【安全違規(guī)】僅允許執(zhí)行 SELECT 查詢 # 加上 LIMIT 1 快速校驗語法與字段合法性 test_sql fSELECT * FROM ({clean_sql}) AS _dry_run_t LIMIT 1; try: cursor self.db.cursor() cursor.execute(test_sql) data cursor.fetchall() return True, data except Exception as e: return False, str(e) def _heal_sql(self, state: SQLCorrectionState) - str: prompt f 你之前生成的 SQL 語句在數(shù)據(jù)庫中執(zhí)行報錯。請根據(jù)以下錯誤信息進行分析并輸出修復后的正確 SQL。 【用戶原始需求】: {state.natural_query} 【可用表結構 Schema】: {state.schema_info} 【出錯的 SQL】: sql {state.current_sql} 【數(shù)據(jù)庫底層返回的真實報錯堆?！? {state.last_error} 【糾錯指引】: 1. 仔細分析報錯原因是字段不存在、缺少 GROUP BY、還是函數(shù)參數(shù)類型不匹配 2. 對照可用表結構替換為正確的字段名或語法 3. 直接輸出修正后的 SQL 語句使用 sql 代碼塊包裹。 response self.llm.generate(prompt) return self._extract_sql_code(response)三、生產(chǎn)調(diào)優(yōu)經(jīng)驗與防御紅線在落地自愈機制時必須嚴守三條鐵律嚴格限定最大自愈輪數(shù)Max Retries ≤ 2實測表明85% 的語法錯誤在第 1 輪自糾錯中就能完美修復另外 10% 在第 2 輪修復超過 2 輪仍未成功的通常屬于 Schema 缺失或邏輯嚴重偏離繼續(xù)循環(huán)只會白白空耗 Token 與延遲。只讀保護與 SQL 注入攔截Read-Only Sandbox自愈引擎執(zhí)行的數(shù)據(jù)庫賬號必須配置為全局只讀用戶SELECT 權限并開啟查詢超時statement_timeout 3000ms杜絕全表鎖死或慢查詢拖垮主庫。只反饋技術報錯不泄露表內(nèi)隱私注入給大模型糾錯的上下文僅包含數(shù)據(jù)庫返回的錯誤信息如Column status in field list is ambiguous嚴禁攜帶任何業(yè)務敏感的行數(shù)據(jù)。用報錯驅(qū)動推理用反饋實現(xiàn)自愈SQL 糾錯循環(huán)讓 Text2SQL 系統(tǒng)真正擁有了從失敗中實時學習、自主修復的工業(yè)級韌性。