# 大模型調用數據庫表實踐：基于自然語言的SQL生成與數據查詢系統

## 一、背景與目標

在企業數據管理場景中，非技術人員（如業務人員、管理人員）常常需要通過數據庫查詢獲取關鍵信息，但直接編寫SQL語句存在技術門檻。傳統的數據庫交互方式依賴專業IT人員，效率較低且響應滯后。本系統通過集成大語言模型（LLM）與數據庫技術，實現**自然語言到SQL語句的自動轉換**，降低數據查詢的技術門檻，提升企業數據利用效率。

## 二、核心技術原理與實現

### 2.1 大模型驅動的SQL生成

系統選用`deepseek-r1:7b`大語言模型作為SQL生成引擎，核心邏輯封裝在`generate_sql`函數中（<mcsymbol name="generate_sql" filename="01.py" path="e:\source\traeproj\test01\ui界面\01.py" startline="29" type="function"></mcsymbol>）。其技術要點包括：

#### （1）Prompt工程設計

通過嚴格的`system prompt`約束模型行為：

- 僅輸出完整SQL語句（以分號結尾），禁止自然語言解釋；

- 強制對齊輸入的表結構（字段名、表名必須與`TABLE_SCHEMA`完全一致）；

- 支持中文表名/字段名（如示例中的"員工表""姓名"）。

用戶側通過`user prompt`傳遞具體需求，包含表結構描述、查詢需求及示例（如："查詢研發部門中工資高于15000元的員工姓名,入職日期,薪資，按入職日期降序排列"），確保生成的SQL符合業務邏輯。

#### （2）生成結果清洗

模型返回內容可能包含Markdown代碼塊（如```sql）或中間思考過程（如`<answer>...`），系統通過正則表達式（`re.sub`）清理冗余信息，最終輸出純凈的SQL語句。具體步驟包括：

- 移除`</answer>...`標簽及內容；

- 清理開頭的```sql標記和結尾的```標記；

- 去除首尾換行符，確保輸出為嚴格SQL語句。

### 2.2 數據庫交互與數據管理

系統基于`SQLAlchemy`實現數據庫操作，支持SQLite輕量級數據庫（<mcsymbol name="initialize_database" filename="01.py" path="e:\source\traeproj\test01\ui界面\01.py" startline="167" type="function"></mcsymbol>），核心功能包括：

#### （1）數據庫初始化

- 檢查`employee.db`是否存在，若不存在則自動創建`員工表`（字段包含`id`、`姓名`、`部門`、`入職日期`、`工資`）；

- 首次初始化時調用`insert_test_data`（<mcsymbol name="insert_test_data" filename="01.py" path="e:\source\traeproj\test01\ui界面\01.py" startline="101" type="function"></mcsymbol>）插入20條模擬數據（使用`Faker`生成中文姓名、部門等信息），并通過`assert`進行類型校驗（如姓名必須為字符串、工資必須為浮點數）。

#### （2）數據查詢執行

`query_database`函數（<mcsymbol name="query_database" filename="01.py" path="e:\source\traeproj\test01\ui界面\01.py" startline="201" type="function"></mcsymbol>）接收生成的SQL語句，通過`pd.read_sql`執行查詢并返回結果（`pandas DataFrame`格式），支持異常捕獲與日志記錄（如SQL為空、數據庫連接失敗等場景）。

## 三、可靠性設計與異常處理

### 3.1 數據質量保障

- **測試數據校驗**：在`insert_test_data`中，每條模擬數據生成時均通過`assert`校驗類型（如`assert isinstance(name, str)`），確保入庫數據符合表結構定義；

- **SQL合法性約束**：通過大模型的`system prompt`強制生成符合表結構的SQL，避免字段名/表名錯誤導致的查詢失敗。

### 3.2 運行時監控與日志

系統集成`logging`模塊（配置于文件頂部），記錄關鍵操作日志：

- 數據插入成功時輸出`INFO`級日志（如"已成功插入20條測試數據"）；

- 數據庫操作異常（如`OperationalError`）、數據類型錯誤（如`AssertionError`）時輸出`ERROR`級日志，并記錄堆棧信息（`exc_info=True`），便于問題定位。

## 四、應用場景與示例演示

### 4.1 典型應用場景

- **企業員工管理**：業務人員通過自然語言查詢"研發部中工資高于15000元的員工"，系統自動生成SQL并返回結果，無需IT人員介入；

- **數據分析輔助**：管理人員通過"近一年入職的市場部員工數量"等自然語言需求，快速獲取統計數據，支撐決策分析。

### 4.2 完整流程示例

以`NATURAL_QUERY`（"查詢研發部門中工資高于15000元的員工姓名,入職日期,薪資，按入職日期降序排列"）為例：

1. **生成SQL**：`generate_sql`函數調用大模型生成SQL（示例輸出：`SELECT 姓名, 入職日期, 工資 FROM 員工表 WHERE 部門 = '研發部' AND 工資 > 15000 ORDER BY 入職日期 DESC;`）；

2. **執行查詢**：`query_database`函數執行上述SQL，返回符合條件的員工數據；

3. **結果輸出**：通過`pandas.to_markdown`以表格形式展示結果（如無數據則提示"無符合條件的數據"）。

# 導入 requests 庫，用于發送 HTTP 請求
import requests
# 從 sqlalchemy 庫導入 create_engine 函數，用于創建數據庫連接引擎
from sqlalchemy import create_engine
# 導入 pandas 庫，用于數據處理和分析
import pandas as pd
# 導入 os 模塊，用于與操作系統進行交互
import os
# 從 sqlalchemy.exc 導入 OperationalError 異常，用于處理數據庫操作錯誤
from sqlalchemy.exc import OperationalError
# 導入 Faker 類，用于生成模擬數據
from faker import Faker  # 新增：用于生成模擬數據
# 導入 random 模塊，用于生成隨機數
import random
# 從 datetime 模塊導入 datetime 和 timedelta 類，用于處理日期和時間
from datetime import datetime, timedelta
# 從 sqlalchemy 導入 text 函數，用于處理 SQL 語句
from sqlalchemy import text  # 新增導入 text 函數
# 導入 logging 模塊，用于日志記錄
import logging  # 新增：日志記錄模塊
# 從 tenacity 庫導入重試相關的類和函數，用于實現重試機制
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
# 從 ollama 庫導入 chat 函數，用于與大語言模型進行交互
from ollama import chat
# 從 ollama 庫導入 ChatResponse 類，用于處理聊天響應
from ollama import ChatResponse# 定義 generate_sql 函數，用于根據自然語言查詢和表結構生成 SQL 語句
def generate_sql(query, table_schema, model="deepseek-r1:7b"):# 調用 chat 函數與大語言模型進行交互，獲取生成的 SQL 語句response: ChatResponse = chat(model, messages=[{"role": "system","content": "你是一個純粹的SQL生成工具，必須嚴格遵循以下規則：\n""1. 僅返回完整的SQL語句（以分號結尾），禁止輸出任何解釋、說明或自然語言文本；\n""2. 嚴格根據輸入的表結構生成語句，字段名和表名必須與表結構完全一致；\n""3. 若輸入中包含中文表名/字段名，直接使用中文，無需翻譯或轉義。"},{'role': 'user','content': f"""1. 表結構：{table_schema}2. 查詢需求：{query}3. 示例："SELECT 姓名, 入職日期 FROM 員工表 WHERE 部門 = '研發部' AND 工資 > 15000 ORDER BY 入職日期 DESC;" （注意：部門字段的值必須為'研發部'，包含'部'字，禁止使用簡寫'研發'）4. 僅輸出sql語句，不允許輸出解釋或說明。條件要與表中的內容保持一致，保證能查出5. 注意初始字段要依據表結構名稱中的字段名一致，不要使用英文字段名"""},],)# 導入 re 模塊，用于正則表達式處理import re# 獲取大語言模型返回的消息內容answer = response['message']['content']# 使用正則表達式移除 <think> 標簽及其內容cleaned = re.sub(r'<think>.*?</think>', '', answer, flags=re.DOTALL)# 清理開頭的 ```sql 標記cleaned = re.sub(r'^\s*```sql\s*', '', cleaned, flags=re.DOTALL)# 清理結尾的 ``` 標記cleaned = re.sub(r'\s*```\s*$', '', cleaned, flags=re.DOTALL)# 移除開頭和結尾的換行符sql = re.sub(r'^\n+|\n+$', '', cleaned)# 打印生成的 SQL 語句print("sql:" + sql)# 返回生成的 SQL 語句return sql# print(response['message']['content'])# or access fields directly from the response object# print(response.message.content)# return response['message']['content']# 配置日志（建議添加在文件頂部）
logging.basicConfig(# 設置日志級別為 INFOlevel=logging.INFO,# 設置日志格式format='%(asctime)s - %(levelname)s - %(message)s',# 設置日志輸出到文件和控制臺handlers=[logging.FileHandler('app.log'), logging.StreamHandler()]
)# 定義 send_messages 函數，用于向大語言模型發送消息并獲取 SQL 語句
def send_messages(messages):# 調用 chat 函數與大語言模型進行交互，獲取生成的 SQL 語句response: ChatResponse = chat(model="deepseek-r1:7b", messages=[{'role': 'user','content': f"""內容：{messages}獲取：返回SQL語句1. 僅返回嚴格的SQL語句（以分號結尾，無任何其他文本）。2. 結果不顯示思考過程文字，僅返回SQL語句。""",},])# 返回聊天響應中的第一條消息return response.choices[0].message# 定義 insert_test_data 函數，用于向數據庫中插入模擬員工數據
def insert_test_data():"""插入20條模擬員工數據（增強版：命名參數綁定+類型校驗）"""# 創建 Faker 實例，用于生成中文模擬數據fake = Faker("zh_CN")# 定義部門列表departments = ["研發部", "市場部", "財務部", "人力資源部"]# 初始化一個空列表，用于存儲模擬數據test_data = []  # 將存儲字典列表（命名參數格式）# 生成數據時增加類型校驗for _ in range(20):# 判斷是否為研發部門數據is_rd = _ < 5# 生成員工姓名并校驗類型name = fake.name()assert isinstance(name, str), "姓名必須為字符串類型"# 生成員工部門并校驗類型department = "研發部" if is_rd else random.choice(departments[1:])assert isinstance(department, str), "部門必須為字符串類型"# 生成員工入職日期并校驗類型hire_date = (datetime.now() - timedelta(days=random.randint(365, 3650))).strftime("%Y-%m-%d")assert isinstance(hire_date, str), "入職日期必須為字符串類型"# 生成員工工資并校驗類型salary = round(random.uniform(12000, 25000) if is_rd else random.uniform(8000, 14000), 2)assert isinstance(salary, float), "工資必須為浮點數類型"# 存儲為字典（命名參數格式）test_data.append({"name": name,"department": department,"hire_date": hire_date,"salary": salary})# 執行插入（改用命名參數綁定）# 創建數據庫連接引擎engine = create_engine(f"sqlite:///employee.db")# 定義插入數據的 SQL 語句，使用命名占位符insert_sql = """INSERT INTO 員工表 (姓名, 部門, 入職日期, 工資) VALUES (:name, :department, :hire_date, :salary)"""try:# 連接數據庫with engine.connect() as conn:# 使用命名參數無需排序，避免類型比較錯誤conn.execute(text(insert_sql), test_data)# 提交事務conn.commit()# 記錄日志，表示數據插入成功logging.info(f"已成功插入{len(test_data)}條測試數據")except OperationalError as e:# 記錄數據庫操作異常日志logging.error(f"數據庫操作異常：{str(e)}，請檢查SQL語句或表結構")except AssertionError as e:# 記錄測試數據類型錯誤日志logging.error(f"測試數據類型錯誤：{str(e)}，請檢查數據生成邏輯")except Exception as e:# 記錄插入測試數據失敗日志，并記錄堆棧信息logging.error(f"插入測試數據失敗：{str(e)}", exc_info=True)# 定義 initialize_database 函數，用于初始化數據庫
def initialize_database():"""檢查數據庫是否存在，不存在則創建表并插入測試數據"""# 定義數據庫文件路徑db_path = "employee.db"  # 固定為當前目錄# 檢查數據庫文件是否存在if not os.path.exists(db_path):# 創建數據庫連接引擎engine = create_engine(f"sqlite:///{db_path}")# 定義創建表的 SQL 語句create_table_sql = f"""CREATE TABLE 員工表 (id INTEGER PRIMARY KEY AUTOINCREMENT,姓名 TEXT,部門 TEXT,入職日期 DATE,工資 REAL);"""try:# 連接數據庫with engine.connect() as conn:# 關鍵修復：使用 text() 包裝 SQL 語句conn.execute(text(create_table_sql))# 打印提示信息，表示數據庫和表已創建print(f"數據庫文件 {db_path} 不存在，已自動創建并初始化員工表")# 插入測試數據（僅首次創建時執行）insert_test_data()except Exception as e:# 打印初始化數據庫失敗的錯誤信息print(f"初始化數據庫失敗：{e}")# 定義 query_database 函數，用于執行 SQL 查詢
def query_database(sql):# 檢查傳入的 SQL 語句是否為空if not sql:# 記錄日志，表示傳入的 SQL 為空，無法執行查詢logging.error("錯誤：傳入的SQL為空，無法執行查詢")  # 替換print為日志return None# 定義數據庫文件路徑db_path = "employee.db"# 創建數據庫連接引擎engine = create_engine(f"sqlite:///{db_path}")try:# 連接數據庫并執行 SQL 查詢with engine.connect() as conn:result = pd.read_sql(sql, conn)# 返回查詢結果return resultexcept Exception as e:# 記錄數據庫查詢失敗的日志，并記錄堆棧信息logging.error(f"數據庫查詢失敗（具體錯誤）：{str(e)}", exc_info=True)return None# 定義 getsql 函數，用于生成 SQL 語句
def getsql():# 調用 generate_sql 函數生成 SQL 語句并返回return generate_sql(NATURAL_QUERY, TABLE_SCHEMA)# 定義表結構
TABLE_SCHEMA = """
員工表(id INTEGER PRIMARY KEY AUTOINCREMENT,姓名 TEXT,部門 TEXT,入職日期 DATE,工資 REAL
)
"""# 自然語言查詢示例
# NATURAL_QUERY = "查詢研發部門中工資高于15000元的員工姓名,入職日期,薪資，按入職日期降序排列"
NATURAL_QUERY = "統計每個部門的工資的平均工資，按照平均工資從多到少排列，輸出部門，平均工資"# 主流程優化
if __name__ == "__main__":# 初始化數據庫initialize_database()# 1. 生成 SQL（增加失敗處理）generated_sql = generate_sql(NATURAL_QUERY, TABLE_SCHEMA)# 檢查生成的 SQL 語句是否為空if not generated_sql:print("生成SQL失敗，終止后續流程")exit(1)# 打印生成的 SQL 語句print("生成的 SQL：\n", generated_sql)# sql = send_messages(generated_sql)# print("sql==="+sql)# 2. 執行查詢result_df = query_database(generated_sql)# 3. 整理并輸出結果（明確錯誤類型）if result_df is None:print("查詢過程中出現錯誤")elif result_df.empty:print("無符合條件的數據。")else:print("\n查詢結果：")print(result_df.to_markdown())  # 以表格形式輸出

## 五、總結

本系統通過大模型與數據庫技術的結合，實現了自然語言到SQL的自動化轉換，降低了數據查詢的技術門檻。核心設計（如嚴格的prompt約束、數據類型校驗、日志監控）保障了系統的可靠性，適用于企業內部數據管理、業務分析等場景，為非技術人員提供了高效的數據交互方式。