纯代码实战--用Deepseek+SQLite+Ollama搭建数据库助手

如何用Python调用本地模型实现DeepSeek提示词模板：一步步教你高效解决13种应用场景
从零到一：纯代码联合PyQt5、Ollama、Deepseek打造简易版智能聊天助手
用外接知识库武装大模型：基于Deepseek、Ollama、LangChain的RAG实战解析
纯代码实战–用Deepseek+SQLite+Ollama搭建数据库助手

概述

本博客手把手教学通过Python调用Deepseek大模型快速搭建智能薪酬问答系统，实现“说人话查数据”。通过整合Ollama大模型与SQLite数据库，开发了一个能理解自然语言、自动生成SQL、安全执行查询并给出人性化解读的AI助手。

文章详解三大核心模块——智能SQL生成器、安全查询引擎、结果解释器，讲解数据流转逻辑，给出大模型连接结构化数据的全链路开发示例。

提示1：完整代码看最后，SQLite示例数据准备后复制代码可直接运行。提示2：本文默认已完成本地Deepseek部署，若未完成见作者另外一篇博客 https://blog.csdn.net/qq_36112576/article/details/145510585?spm=1001.2014.3001.5501

SQLite示例数据准备

复制以下代码运行，生成数据库example.db，其中表名employees；属性名id,name, position, salary。

import sqlite3# 连接到 SQLite 数据库（如果数据库文件不存在，会自动创建一个新的数据库）
conn = sqlite3.connect("example.db")# 创建一个游标对象，通过游标来执行 SQL 语句
cursor = conn.cursor()# 1. 创建一个示例表（如果表不存在）
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (id INTEGER PRIMARY KEY,name TEXT NOT NULL,position TEXT,salary REAL
)
''')# 2. 插入一些示例数据（如果表为空）
cursor.execute("INSERT INTO employees (id,name, position, salary) VALUES (?,?, ?, ?)", (12100,'Alice', 'Manager', 80000.0))
cursor.execute("INSERT INTO employees (id,name, position, salary) VALUES (?,?, ?, ?)", (12566,'Bob', 'Developer', 60000.0))
cursor.execute("INSERT INTO employees (id,name, position, salary) VALUES (?,?, ?, ?)", (10465,'Charlie', 'Designer', 50000.0))# 提交事务，保存数据
conn.commit()# 3. 查询数据并打印出来
cursor.execute("SELECT * FROM employees")
rows = cursor.fetchall()print("查询结果:")
for row in rows:print(f"ID: {row[0]}, Name: {row[1]}, Position: {row[2]}, Salary: {row[3]}")# 4. 根据条件查询数据，例如查询所有薪水高于 60000 的员工
cursor.execute("SELECT * FROM employees WHERE salary > ?", (60000,))
rows = cursor.fetchall()print("\n薪水大于 60000 的员工:")
for row in rows:print(f"ID: {row[0]}, Name: {row[1]}, Position: {row[2]}, Salary: {row[3]}")# 5. 根据员工姓名查询某个员工的信息
cursor.execute("SELECT * FROM employees WHERE name = ?", ('Bob',))
row = cursor.fetchone()if row:print(f"\n查询到的员工信息: ID: {row[0]}, Name: {row[1]}, Position: {row[2]}, Salary: {row[3]}")
else:print("\n未找到该员工信息。")# 6. 关闭游标和连接
cursor.close()
conn.close()

根据“cursor.execute("INSERT INTO employees ****”自己修改数据库，以下实验使用数据库实例：
在这里插入图片描述

整体流程

数据流转7步：

（1）用户->>+系统: 输入自然语言问题

question = "工资高于60000的人有哪些"

（2）系统->>+Ollama: 发送生成SQL请求

generate_sql_stream(question):

（3）Ollama–>>-系统: 流式返回SQL

elif status == 'complete':sql = contentprint(f"\n\n✅ 生成 SQL 成功：\n{sql}")

（4）系统->>+SQLite: 执行安全查询

results = execute_sql_query(sql)

（5）SQLite–>>-系统: 返回数据集

print(f"找到 {len(results)} 条相关记录")

（6）系统->>+Ollama: 发送结果解释请求

generate_answer_stream(question, results)

（7）Ollama–>>-用户: 流式输出自然语言结果

print(chunk, end='', flush=True)
answer_buffer.append(chunk)

执行显示示例

🚀 正在分析问题...
SELECT name, position, salary FROM employees WHERE salary > 60000 
✅ 生成 SQL 成功🔍 正在查询数据库...
找到 8 条相关记录📝 正在生成回答：
根据系统查询，薪资超过6万元的员工共有8人，其中：
- 技术总监张三的薪资为85000元
- 首席架构师李四...

核心代码讲解

1、SQL生成引擎（核心安全设计）

def generate_sql_stream(question: str):...# 关键prompt设计prompt = f"""您是人资系统专用SQL助手，请根据员工表结构生成查询：表结构：id(工号), name(姓名), position(职位), salary(薪资)要求：仅返回SELECT语句，包含安全过滤条件问题：{question}"""# 流式生成控制full_response = []for response in ollama.chat(model=MODEL_NAME,messages=messages,stream=True,options={'temperature': 0.2}):chunk = response['message']['content']full_response.append(chunk)yield ('loading', chunk)# 安全过滤（双重校验）final_sql = final_sql.replace('```sql','').strip()  # 去除代码标识if not final_sql.lower().startswith('select'):raise BlockedOperationErrorif not any(op in final_sql_lower for op in [' where ', ' like ', ' = ']):raise QueryGenerationError("缺少必要过滤条件")

设计：

Prompt工程限定生成范围；
流式输出降低等待焦虑；
正则清洗+语法校验双保险。

在这里插入图片描述

完整代码中涉及到大量的sql返回语句格式修正，如下，主要是Deepseek返回结果为final_sql ，它包括完整的think过程，并不是直接可用的SQL，若SQL语句生成失败，对这部分内容进行debug。

        final_sql = ''.join(full_response).strip()sql_start = final_sql.find("</think>") + 9sql_end = final_sql.find(";", sql_start) + 1final_sqlout = final_sql[sql_start:sql_end].strip()# 安全验证if SAFETY_CHECK:final_sql_lower = final_sqlout.lower()# 去除可能存在的代码标识符（例如```sql）final_sql_lower = final_sql_lower.replace('```sql', '').replace('```', '').strip()

2、数据库查询执行器（安全防线）

def execute_sql_query(sql: str):# 危险操作黑名单blocked_keywords = ['insert', 'update', 'delete', 'drop']if any(kw in sql.lower() for kw in blocked_keywords):raise SecurityAlert("危险操作拦截")# 上下文管理器保安全with sqlite3.connect(DATABASE_PATH) as conn:return conn.execute(sql).fetchall()

设计：

SQL注入关键词过滤，防止增删改等危险；
自动关闭数据库连接。

在这里插入图片描述

3、回答生成器

根据检索内容回答问题，示例结果如下：
在这里插入图片描述

def generate_answer_stream(question, data):# 数据结构化预处理context = "查询结果：\n" + "\n".join(f"{name} | {position} | ￥{salary:,}"for _, name, position, salary in data)# 生成提示词模板messages = [{"role": "system", "content": "您是人资专员，请将数据转换为自然语言回答"},{"role": "user", "content": f"问题：{question}\n{context}"}]# 流式生成回答for chunk in ollama.chat(...):yield chunk

这里容易出现未预期错误：tuple index out of range，这是由于第一步SQL生成的查询语句并不固定，以下都有可能，导致context内容不固定，需要通过提示词进一步规划SQL语句生成，暂时未优化，读者修改后可以评论区把优化语句打出来。

select name, id, position from employees where salary > 60000;
select name, salary from employees where salary > 60000;
select name, id, position,salary from employees where salary > 60000;

完整代码

注意：需要先准备好数据库example.db，见前文SQLite示例数据准备。

import sqlite3
import ollama
from typing import List, Tuple, Optional
from time import sleep# 配置参数
MODEL_NAME = 'deepseek-r1:7b'  # 本地运行的 Ollama 模型名称
DATABASE_PATH = 'example.db'  # 数据库路径
SAFETY_CHECK = True  # 是否启用安全校验class QueryGenerationError(Exception):"""SQL 生成异常"""passclass DatabaseError(Exception):"""数据库操作异常"""passdef generate_sql_stream(question: str) -> str:"""流式生成 SQL 查询语句（带安全验证）返回格式: (status, content)status: loading|complete|error"""prompt = f"""您是一个专业的人力薪酬数据库查询助手。根据用户问题生成安全且准确的 SQL 查询。数据库表结构：
- 表名：employees 
- 字段：
列名: id, 类型: INTEGER
列名: name, 类型: TEXT
列名: position, 类型: TEXT
列名: salary, 类型: REAL当前问题：{question}生成要求：
1. 仅限 SELECT 查询
2. 输出纯 SQL 不带 Markdown"""messages = [{'role': 'system', 'content': '您是一个 SQL 生成专家，只返回有效的 SELECT 语句'},{'role': 'user', 'content': prompt}]try:full_response = []for response in ollama.chat(model=MODEL_NAME,messages=messages,stream=True,options={'temperature': 0.2}):chunk = response['message']['content']full_response.append(chunk)yield ('loading', chunk)final_sql = ''.join(full_response).strip()sql_start = final_sql.find("</think>") + 9sql_end = final_sql.find(";", sql_start) + 1final_sqlout = final_sql[sql_start:sql_end].strip()# 安全验证if SAFETY_CHECK:final_sql_lower = final_sqlout.lower()# 去除可能存在的代码标识符（例如```sql）final_sql_lower = final_sql_lower.replace('```sql', '').replace('```', '').strip()#print(final_sql_lower)if not final_sql_lower.startswith('select'):raise QueryGenerationError("禁止非 SELECT 操作")# if ';' in final_sql:#     raise QueryGenerationError("检测到多语句查询")# if not any(op in final_sql_lower for op in [' where ', ' like ', ' = ']):#     raise QueryGenerationError("缺少必要过滤条件")yield ('complete', final_sql_lower)except Exception as e:yield ('error', f"SQL 生成失败: {str(e)}")def execute_sql_query(sql: str) -> List[Tuple]:"""执行 SQL 查询并返回结果"""try:conn = sqlite3.connect(DATABASE_PATH)cursor = conn.cursor()# 预验证sql_lower = sql.lower()if any(keyword in sql_lower for keyword in ['insert', 'update', 'delete', 'drop']):raise DatabaseError("危险操作被拦截")cursor.execute(sql)results = cursor.fetchall()return resultsexcept sqlite3.Error as e:raise DatabaseError(f"数据库错误: {str(e)}")finally:if 'conn' in locals():conn.close()def generate_answer_stream(question: str, data: List[Tuple]) -> str:"""流式生成自然语言回答"""if not data:yield "未找到相关记录"returncontext = "name：\n" + '\n'.join([f"- name：{row[1]}（salary：{row[3]}）" for row in data])messages = [{'role': 'system', 'content': '您是一个薪酬系统管理助理，用正式中文回答客户问题'},{'role': 'user', 'content': f"问题：{question}\n\n{context}"}]try:buffer = []for response in ollama.chat(model=MODEL_NAME,messages=messages,stream=True,options={'temperature': 0.7}):content = response['message']['content']buffer.append(content)yield content# 后处理验证final_answer = ''.join(buffer)if len(final_answer) < 20:raise RuntimeError("回答过短，可能生成失败")except Exception as e:yield f"\n回答生成异常：{str(e)}"def main():try:# 用户问题question = "工资高于60000的人有哪些"# 阶段1：生成 SQLprint("🚀 正在分析问题...")sql = Nonefor status, content in generate_sql_stream(question):if status == 'loading':print(content, end='', flush=True)sleep(0.02)  # 模拟流式效果elif status == 'complete':sql = contentprint(f"\n\n✅ 生成 SQL 成功：\n{sql}")elif status == 'error':raise QueryGenerationError(content)# 阶段2：执行查询print("\n🔍 正在查询数据库...")results = execute_sql_query(sql)print(f"找到 {len(results)} 条相关记录")# 阶段3：生成回答print("\n📝 正在生成回答：")answer_buffer = []for chunk in generate_answer_stream(question, results):print(chunk, end='', flush=True)answer_buffer.append(chunk)# 保存完整回答full_answer = ''.join(answer_buffer)print(f"\n\n💡 完整回答已生成，可保存至日志文件")except QueryGenerationError as e:print(f"\n❌ SQL 生成错误：{str(e)}")except DatabaseError as e:print(f"\n❌ 数据库错误：{str(e)}")except Exception as e:print(f"\n❌ 未预期错误：{str(e)}")if __name__ == "__main__":main()