SQL 语法纠错循环：基于数据库执行报错的 Agent 动态自愈
在 Text2SQL 的真实生产落地中，即便前置完成了精准的 Schema 召回与 Few-Shot 注入，大模型一次性写出 100% 完美无瑕 SQL 的概率依然很难突破 80%。
大模型生成的 SQL 常常会遭遇各种数据库底层的刚性报错：
- 字段或表名拼写错误（如误写为
user_name而实际列名为username）； - 聚合函数与 GROUP BY 缺失（如在
SELECT dept_id, count(*)中遗漏了GROUP BY dept_id）； - 数据类型隐式转换失败（如将字符串与时间戳直接做大于等于比较）；
- 方言特有语法不兼容（在 ClickHouse 中误用了 MySQL 的专有函数）。
如果系统在遇到数据库报错时直接把错误信息抛给最终用户，整个系统的可用性将大打折扣。构建一个**“生成 -> 预执行 -> 捕获报错 -> 注入反思 -> 动态自纠错（Self-Correction Loop）”**的闭环机制，是保障 Text2SQL 端到端执行成功率突破 95% 的核心技术底牌。
一、动态自愈闭环的执行流架构
[ 用户自然语言 Query ] │ ▼ ┌─────────────────────────────────┐ │ 1. 初次 SQL 生成 (Initial Plan) │ └────────────────┬────────────────┘ │ ▼ ┌─────────────────────────────────┐ │ 2. 只读沙箱执行 (Dry-Run Guard) │ ──(带有 LIMIT 1 与只读事务保护) └────────────────┬────────────────┘ │ ┌────────┴────────┐ ▼ ▼ [ 执行成功 (Success) ] [ 捕获 DB 异常 (Error Traced) ] 格式化输出业务结论 提取精准错误码与详细堆栈 │ ▼ ┌─────────────────────────────────┐ │ 3. 错误提示词组装与上下文增强 │ │ (结合原始 SQL + 报错原因 + Schema)│ └────────────────┬────────────────┘ │ ▼ ┌─────────────────────────────────┐ │ 4. 大模型自纠错 (Re-generate) │ ──(最大重试 2 轮) └─────────────────────────────────┘二、生产级自纠错 Agent 的 Python 实现
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_query=query, schema_info=schema_info, current_sql=initial_sql ) while state.retry_count <= state.max_retries: # 2. 尝试在数据库中执行验证 (带有只读拦截) 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. 记录报错并触发自愈推理 state.last_error = result_or_err state.retry_count += 1 if state.retry_count > state.max_retries: break # 超过最大自愈次数，终止 # 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]: """安全预执行：限制只读与返回条数""" clean_sql = sql.strip().rstrip(";") if not clean_sql.upper().startswith("SELECT"): return False, "【安全违规】仅允许执行 SELECT 查询" # 加上 LIMIT 1 快速校验语法与字段合法性 test_sql = f"SELECT * 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 语句在数据库中执行报错。请根据以下错误信息进行分析并输出修复后的正确 SQL。 【用户原始需求】: {state.natural_query} 【可用表结构 Schema】: {state.schema_info} 【出错的 SQL】: ```sql {state.current_sql} ``` 【数据库底层返回的真实报错堆栈】: {state.last_error} 【纠错指引】: 1. 仔细分析报错原因是字段不存在、缺少 GROUP BY、还是函数参数类型不匹配； 2. 对照可用表结构，替换为正确的字段名或语法； 3. 直接输出修正后的 SQL 语句，使用 ```sql 代码块包裹。 """ response = self.llm.generate(prompt) return self._extract_sql_code(response)三、生产调优经验与防御红线
在落地自愈机制时，必须严守三条铁律：
- 严格限定最大自愈轮数（Max Retries ≤ 2）：
实测表明，85% 的语法错误在第 1 轮自纠错中就能完美修复，另外 10% 在第 2 轮修复；超过 2 轮仍未成功的，通常属于 Schema 缺失或逻辑严重偏离，继续循环只会白白空耗 Token 与延迟。 - 只读保护与 SQL 注入拦截（Read-Only Sandbox）：
自愈引擎执行的数据库账号必须配置为全局只读用户（SELECT 权限），并开启查询超时（statement_timeout = 3000ms），杜绝全表锁死或慢查询拖垮主库。 - 只反馈技术报错，不泄露表内隐私：
注入给大模型纠错的上下文仅包含数据库返回的错误信息（如Column 'status' in field list is ambiguous），严禁携带任何业务敏感的行数据。
用报错驱动推理，用反馈实现自愈，SQL 纠错循环让 Text2SQL 系统真正拥有了从失败中实时学习、自主修复的工业级韧性。
