【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}")