【GenBI优化】提升text2sql准确率：建议使用推理大模型，增加重试

引言

Text-to-SQL（文本转 SQL）是自然语言处理（NLP）领域的一项重要任务，旨在将自然语言问题自动转换为可在数据库上执行的 SQL 查询语句。这项技术在智能助手、数据分析工具、商业智能（BI）平台等领域具有广泛的应用前景，能够极大地降低数据查询和分析的门槛，让非技术用户也能轻松地与数据交互。

近年来，随着深度学习和预训练语言模型（PLM）的快速发展，Text-to-SQL 模型的性能取得了显著提升。然而，在实际应用中，Text-to-SQL 模型的准确率仍然面临诸多挑战，例如自然语言的多样性和歧义性、复杂 SQL 查询的处理、数据库 Schema 的理解和利用等。

本文将深入探讨两种有效提升 Text-to-SQL 准确率的方法：

使用推理大模型（Large Language Models for Inference）： 利用更大规模、更强大的预训练语言模型进行推理，充分发挥其语义理解和生成能力。
增加重试机制（Retry Mechanism）： 在模型生成 SQL 查询的过程中，引入重试机制，对生成的 SQL 进行验证和修正，提高最终结果的准确性。

我们将详细介绍这两种方法的原理、实现方式，并通过丰富的代码示例和表格进行说明，帮助读者深入理解并掌握这些技术。

1. 使用推理大模型 (Large Language Models for Inference)

1.1. 为什么选择推理大模型？

传统的 Text-to-SQL 模型通常基于较小规模的预训练语言模型（例如 BERT、RoBERTa）进行微调。虽然这些模型在许多基准数据集上取得了不错的效果，但在处理复杂、多样化的自然语言查询时，其性能往往会受到限制。

近年来，随着模型规模的不断扩大，涌现出了一批参数量巨大的预训练语言模型，例如 GPT-3、Codex、PaLM、LLaMA 等。这些大模型在海量数据上进行预训练，学习了更丰富的语言知识和世界知识，具有更强的语义理解、推理和生成能力。

时间来到当下，逻辑能力最强的是推理模型，DeepSeek R1、openai o1等

使用推理大模型进行 Text-to-SQL 任务具有以下优势：

更强的语义理解能力： 大模型能够更好地理解自然语言查询的意图，捕捉细微的语义差别，处理复杂的表达方式。
更强的泛化能力： 大模型在海量数据上进行预训练，具有更强的泛化能力，能够处理未见过的查询类型和领域知识。
更强的生成能力： 大模型能够生成更流畅、更符合语法规则的 SQL 查询语句。
Zero-shot 或 Few-shot 能力： 一些大模型展现出在少量样本甚至零样本情况下进行 Text-to-SQL 的能力，降低了对大规模标注数据的依赖。

1.2. 推理大模型的选择

选择合适的推理大模型对于提升 Text-to-SQL 准确率至关重要。

选择推理大模型时，需要考虑以下因素：

模型性能： 不同模型在 Text-to-SQL 任务上的性能可能有所差异，需要根据具体数据集和任务进行评估。
模型规模： 模型规模越大，通常性能越好，但计算资源消耗也越大。
API 接口： 一些模型提供了 API 接口，可以方便地进行调用，而另一些模型需要自行部署。
成本： 使用 API 接口可能需要付费，需要考虑成本因素。
开源与否： 开源模型可以自由使用和修改，但可能需要自行搭建环境和进行部署。

1.3. 使用推理大模型进行 Text-to-SQL 的方法

使用推理大模型进行 Text-to-SQL 任务，通常有以下几种方法：

直接生成 SQL（Direct Generation）： 将自然语言查询和数据库 Schema 信息作为输入，直接让模型生成 SQL 查询语句。
基于 Prompt 的生成（Prompt-based Generation）： 设计合适的 Prompt（提示），引导模型生成正确的 SQL 查询。
Few-shot Learning： 提供少量示例（Prompt 中包含示例），让模型学习如何将自然语言查询转换为 SQL 查询。
Zero-shot Learning： 不提供任何示例，直接让模型根据自然语言查询和 Schema 信息生成 SQL 查询（需要模型具有强大的 zero-shot 能力）。

1.4. 代码示例：使用 DeepSeek R1进行 Text-to-SQL

Python 代码示例：

import requests
def text_to_sql(query, schema):"""Args:query: 自然语言查询。schema: 数据库 Schema 信息（字符串形式）。Returns:生成的 SQL 查询（字符串形式），如果生成失败则返回 None。"""prompt = f"""数据库 Schema：{schema}自然语言查询：{query}请生成对应的 SQL 查询：```sql"""DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"  # Example endpointDEEPSEEK_API_KEY = ""  # Your DeepSeek API keyheaders = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}","Content-Type": "application/json"}payload = {"model": "deepseek-reasoner",  # deepseek r1"messages": [{"role": "system", "content": ""},{"role": "user", "content": prompt}],"max_tokens": 10,"temperature": 0}response = requests.post(DEEPSEEK_API_URL, headers=headers, json=payload)# Parse the responseif response.status_code == 200:result = response.json()return result["choices"][0]["message"]["content"].strip()else:print(f"Error calling DeepSeek API: {response.status_code} - {response.text}")return None# 示例数据库 Schema
schema = """
CREATE TABLE Customers (CustomerID INT PRIMARY KEY,Name VARCHAR(255),City VARCHAR(255)
);CREATE TABLE Orders (OrderID INT PRIMARY KEY,CustomerID INT,OrderDate DATE,Amount DECIMAL(10, 2),FOREIGN KEY (CustomerID) REFERENCES Customers(CustomerID)
);
"""# 示例自然语言查询
query = "Find the total amount of orders for each customer in New York."# 调用 text_to_sql 函数生成 SQL 查询
sql_query = text_to_sql(query, schema)if sql_query:print(f"生成的 SQL 查询：\n{sql_query}")

注意：

可以加入few shot示例，比如：

prompt = f"""数据库 Schema：{schema}自然语言查询：列出所有顾客的名字SQL查询：```sqlSELECT Name FROM Customers;```自然语言查询：{query}请生成对应的 SQL 查询：```sql"""

1.5. 实验结果对比(仅供参考)
使用不同模型进行Text-to-SQL任务，并进行对比. 假设使用Spider数据集进行测试。

模型	执行准确率 (EM)	备注
BERT-base	~50-55%	经典预训练语言模型
RoBERTa-large	~60-65%	BERT 的改进版
RAT-SQL + RoBERTa	~70-75%	结合 Schema 信息
GPT-3.5 (few-shot)	~75-80%	使用少量示例
GPT-4 (few-shot)	~80-85%	使用少量示例

结果分析：

可以看到，随着模型规模的增大和技术的进步，Text-to-SQL 模型的准确率不断提升。
使用推理大模型（例如 GPT-3.5、GPT-4）可以显著提高 Text-to-SQL 的准确率。
即使是 few-shot learning（提供少量示例），大模型也能取得很好的效果。

2. 增加重试机制 (Retry Mechanism)

2.1. 为什么需要重试机制？

尽管使用推理大模型可以显著提高 Text-to-SQL 的准确率，但模型仍然可能生成错误的 SQL 查询，原因可能包括：

模型自身的局限性： 即使是最大的模型，也无法保证 100% 的准确率。
自然语言的歧义性： 自然语言查询可能存在多种解释，模型可能选择了错误的解释。
复杂的 SQL 语法： 生成复杂的 SQL 查询（例如涉及嵌套查询、聚合函数、窗口函数等）更容易出错。
输入错误： 用户的输入拼写错误，或者数据库的schema描述存在错误。

为了进一步提高 Text-to-SQL 的准确率，我们可以引入重试机制。重试机制的基本思想是：

生成多个候选 SQL 查询： 让模型一次性生成多个候选 SQL 查询。
验证 SQL 查询： 对生成的 SQL 查询进行验证，例如检查其语法是否正确、是否可以在数据库上执行、执行结果是否符合预期等。
选择最佳 SQL 查询： 从多个候选 SQL 查询中选择最佳的 SQL 查询作为最终结果。
多次尝试： 如果所有候选查询都不符合预期，可以进行多次尝试。

2.2. 重试机制的实现方式

重试机制可以有多种实现方式，以下是一些常用的方法：

基于语法检查的重试 (Syntax-based Retry)：
- 使用 SQL 解析器（例如 sqlparse）检查生成的 SQL 查询是否符合语法规则。
- 如果语法错误，则重新生成 SQL 查询，或者对错误的 SQL 进行修正。
基于执行结果的重试 (Execution-based Retry)：
- 在数据库上执行生成的 SQL 查询。
- 如果执行失败（例如语法错误、表或列不存在等），则重新生成 SQL 查询。
- 如果执行成功，但结果为空或不符合预期（例如返回的行数太少、结果与常识不符等），也可以考虑重新生成 SQL 查询。
基于置信度的重试 (Confidence-based Retry)：
- 让模型对生成的每个 SQL 查询给出一个置信度分数（例如，使用 softmax 概率）。
- 如果置信度低于某个阈值，则重新生成 SQL 查询。
基于多样性的重试 (Diversity-based Retry)：
- 在生成多个候选 SQL 查询时，鼓励模型生成多样化的查询，避免生成多个相似的查询。
- 可以使用不同的解码策略（例如 beam search、sampling）或在 Prompt 中添加多样性相关的指令。
基于反馈的重试 (Feedback-based Retry)：
- 如果模型生成的 SQL 查询不正确，可以向模型提供反馈信息，例如错误类型、错误位置等，帮助模型修正错误。
多次尝试:
- 设置最大尝试次数.
- 每次尝试时, 可以稍微修改Prompt, 或者使用不同的解码策略.

2.3. 代码示例：基于语法检查和执行结果的重试

以下是一个结合语法检查和执行结果验证的重试机制的 Python 代码示例（假设你已经有一个 text_to_sql 函数，可以将自然语言查询转换为 SQL 查询）：

import sqlparse
import sqlite3  # 或其他数据库连接库def text_to_sql_with_retry(query, schema, db_path, max_retries=3):"""带重试机制的 Text-to-SQL 函数。Args:query: 自然语言查询。schema: 数据库 Schema 信息（字符串形式）。db_path: 数据库文件路径。max_retries: 最大重试次数。Returns:生成的 SQL 查询（字符串形式），如果生成失败则返回 None。"""for i in range(max_retries):# 1. 生成 SQL 查询sql_query = text_to_sql(query, schema) # 假设你已经有一个 text_to_sql 函数if sql_query is None:continue# 2. 语法检查parsed = sqlparse.parse(sql_query)if not parsed or parsed[0].get_type() == 'UNKNOWN':print(f"Retry {i+1}: Syntax error in SQL query: {sql_query}")continue# 3. 执行结果验证try:conn = sqlite3.connect(db_path)  # 连接数据库cursor = conn.cursor()cursor.execute(sql_query)results = cursor.fetchall()  # 获取所有结果conn.close()# 简单检查：结果不能为空if results:print(f"Success after {i+1} retries.")return sql_queryelse:print(f"Retry {i+1}: Empty result for SQL query: {sql_query}")except Exception as e:print(f"Retry {i+1}: Execution error for SQL query: {sql_query}\nError: {e}")print("Max retries reached. Failed to generate a valid SQL query.")return None# 示例用法 (假设你有一个名为 "mydatabase.db" 的 SQLite 数据库)
db_path = "mydatabase.db"
query = "Show the names of customers who placed orders after 2023-01-01."
schema = """...""" # 与前面示例相同sql_query = text_to_sql_with_retry(query, schema, db_path)if sql_query:print(f"最终生成的 SQL 查询：\n{sql_query}")

代码解析：

导入必要的库：
- sqlparse: 用于 SQL 语法检查。
- sqlite3 (或其他数据库连接库): 用于连接数据库并执行 SQL 查询。
定义 text_to_sql_with_retry 函数：
- 接收自然语言查询、Schema 信息、数据库路径和最大重试次数作为输入。
- 使用 for 循环进行多次尝试。
生成 SQL 查询：
- 调用 text_to_sql 函数生成 SQL 查询 (你需要根据前面的示例实现 text_to_sql 函数)。
语法检查：
- 使用 sqlparse.parse() 解析生成的 SQL 查询。
- 检查解析结果是否有效，以及第一个语句的类型是否为 UNKNOWN (表示解析失败)。
- 如果语法错误，打印错误信息，并进行下一次重试。
执行结果验证：
- 连接数据库。
- 执行 SQL 查询。
- 获取所有查询结果。
- 检查结果是否为空。如果为空，打印信息并进行下一次重试。
- 如果执行过程中发生异常 (例如数据库连接错误、SQL 执行错误)，打印错误信息并进行下一次重试。
返回结果：
- 如果成功生成有效的 SQL 查询并获取到非空结果，则返回该 SQL 查询。
- 如果达到最大重试次数仍未成功，则返回 None。

注意：

这个示例使用了 SQLite 数据库，你可以根据需要修改为其他数据库 (例如 MySQL, PostgreSQL)。
示例中的执行结果验证只做了简单的非空检查，你可以根据实际需求添加更复杂的验证逻辑，例如检查结果的类型、数量、是否符合预期等。
可以结合多种重试机制，例如先进行语法检查，再进行执行结果验证，最后进行基于置信度的重试。
可以调整最大重试次数。
可以在每次重试时，稍微修改Prompt的内容，或者使用不同的模型生成参数，以提高生成多样性。

2.4 示例: 多次尝试及修改Prompt

import openai
import sqlparse
import sqlite3def text_to_sql_with_retry_and_prompt_variation(query, schema, db_path, max_retries=3):"""带重试机制和 Prompt 变化的 Text-to-SQL 函数。Args:query: 自然语言查询。schema: 数据库 Schema 信息（字符串形式）。db_path: 数据库文件路径。max_retries: 最大重试次数。Returns:生成的 SQL 查询（字符串形式），如果生成失败则返回 None。"""base_prompt = f"""数据库 Schema：{schema}自然语言查询：{query}请生成对应的 SQL 查询：```sql"""for i in range(max_retries):# 根据重试次数修改 Promptif i == 0:prompt = base_promptelif i == 1:prompt = base_prompt + "\n请确保 SQL 查询的语法正确。"else:prompt = base_prompt + "\n请使用最有效的方式查询, 并确保 SQL 查询的语法正确。"# 生成 SQL 查询try:response = openai.Completion.create(engine="text-davinci-003",prompt=prompt,max_tokens=200,n=1,stop=["```"],temperature=0.7 + (i * 0.1),  # 每次重试稍微提高温度)sql_query = response.choices[0].text.strip()except Exception as e:print(f"Retry {i + 1}: API call failed: {e}")continueif sql_query is None:continue# 语法检查parsed = sqlparse.parse(sql_query)if not parsed or parsed[0].get_type() == 'UNKNOWN':print(f"Retry {i+1}: Syntax error in SQL query: {sql_query}")continue# 执行结果验证try:conn = sqlite3.connect(db_path)cursor = conn.cursor()cursor.execute(sql_query)results = cursor.fetchall()conn.close()if results:print(f"Success after {i+1} retries.")return sql_queryelse:print(f"Retry {i+1}: Empty result for SQL query: {sql_query}")except Exception as e:print(f"Retry {i+1}: Execution error for SQL query: {sql_query}\nError: {e}")print("Max retries reached. Failed to generate a valid SQL query.")return None# 示例用法
db_path = "mydatabase.db"
query = "Show the names of customers who placed orders after 2023-01-01."
schema = """..."""  # 与前面示例相同
sql_query = text_to_sql_with_retry_and_prompt_variation(query, schema, db_path)
if sql_query:print(f"最终生成的 SQL 查询：\n{sql_query}")