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_queryquery, schema_infoschema_info, current_sqlinitial_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 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 语句在数据库中执行报错。请根据以下错误信息进行分析并输出修复后的正确 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 系统真正拥有了从失败中实时学习、自主修复的工业级韧性。
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_queryquery, schema_infoschema_info, current_sqlinitial_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 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 语句在数据库中执行报错。请根据以下错误信息进行分析并输出修复后的正确 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 系统真正拥有了从失败中实时学习、自主修复的工业级韧性。
