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