大模型调用数据库表实践：基于自然语言的SQL生成与数据查询系统

# 大模型调用数据库表实践：基于自然语言的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约束、数据类型校验、日志监控）保障了系统的可靠性，适用于企业内部数据管理、业务分析等场景，为非技术人员提供了高效的数据交互方式。