Python调用DeepSeek API实现对本地数据库的AI管理
更新时间：2025年02月05日 09:56:15 作者：mosquito_lover1
这篇文章主要为大家详细介绍了Python如何基于DeepSeek模型实现对本地数据库的AI管理,文中的示例代码简洁易懂,有需要的小伙伴可以跟随小编一起学习一下
场景描述
基于DeepSeek模型，实现对本地数据库的AI管理。
实现思路
1、本地python+flask搭建个WEB，配置数据源。
2、通过DeepSeek模型根据用户输入的文字需求，自动生成SQL语句。
3、通过SQL执行按钮，实现对数据库的增删改查。
前置条件
到DeepSeek官网的API开放平台注册，完成以下配置：
DEEPSEEK_API_KEY=your-deepseek-api-key
DEEPSEEK_API_URL=https://api.deepseek.com/v1/chat
效果展示
核心代码
from flask import Blueprint, render_template, request, jsonify, current_app
from .database import DatabaseManager
from .config import DatabaseConfig
import requests
from .sql_generator import SQLGenerator # 添加这行导入
main = Blueprint('main', __name__)
db_manager = DatabaseManager()
@main.route('/')
def index():
return render_template('index.html')
@main.route('/connect', methods=['POST'])
def connect_database():
try:
data = request.json
print(f"收到连接请求: {data}") # 添加调试信息
# 先创建一个没有指定数据库的连接
config = DatabaseConfig(
host=data.get('host'),
user=data.get('user'),
password=data.get('password'),
database='', # 先不指定数据库
port=int(data.get('port', 3306))
)
success, error_message = db_manager.connect(config)
if success:
# 获取数据库列表
databases = db_manager.get_databases()
print(f"成功获取数据库列表: {databases}") # 添加调试信息
return jsonify({
'success': True,
'databases': databases
})
print(f"连接失败: {error_message}") # 添加调试信息
return jsonify({
'success': False,
'message': error_message
})
except Exception as e:
error_message = f"发生错误: {str(e)}"
print(error_message) # 添加调试信息
import traceback
print(traceback.format_exc()) # 打印完整的错误堆栈
return jsonify({
'success': False,
'message': error_message
})
@main.route('/select-database', methods=['POST'])
def select_database():
try:
data = request.json
database = data.get('database')
if not database:
return jsonify({
'success': False,
'message': '请选择数据库'
})
print(f"切换到数据库: {database}")
# 使用选择的数据库重新连接
config = DatabaseConfig(
host=db_manager.host,
user=db_manager.user,
password=db_manager.password,
database=database,
port=db_manager.port
)
success, error_message = db_manager.connect(config)
if success:
print(f"成功切换到数据库: {database}")
return jsonify({
'success': True,
'message': f'成功切换到数据库: {database}'
})
else:
print(f"切换数据库失败: {error_message}")
return jsonify({
'success': False,
'message': f'切换数据库失败: {error_message}'
})
except Exception as e:
error_message = f"选择数据库时发生错误: {str(e)}"
print(error_message)
import traceback
print(traceback.format_exc())
return jsonify({
'success': False,
'message': error_message
})
@main.route('/generate-sql', methods=['POST'])
def generate_sql():
try:
user_input = request.json.get('input')
model_type = request.json.get('model') # 默认值
print(f"收到SQL生成请求，用户输入: {user_input}，选择模型: {model_type}")
if model_type == '':
# 使用本地 SQL 生成器
generated_sql = SQLGenerator.generate_sql(user_input)
return jsonify({
'success': True,
'sql': generated_sql
})
else:
# 使用其他模型的现有逻辑
model_config = current_app.config['AVAILABLE_MODELS'].get(model_type)
if not model_config:
return jsonify({
'success': False,
'message': '不支持的模型类型'
})
# 通用提示词
prompt = f"""
作为一个SQL专家，请根据以下需求生成合适的SQL语句：
需求：{user_input}
请只返回SQL语句，不需要其他解释。
如果是创建表，请包含合适的字段类型和必要的约束。
"""
try:
headers = {
'Authorization': f'Bearer {model_config["api_key"]}',
'Content-Type': 'application/json'
}
if model_type == 'deepseek':
payload = {
'model': model_config['model_name'],
'messages': [
{'role': 'user', 'content': prompt}
],
'temperature': 0.3
}
else: # OpenAI
payload = {
'model': model_config['model_name'],
'messages': [
{'role': 'system', 'content': 'You are a SQL expert. Only return SQL statements without any explanation.'},
{'role': 'user', 'content': prompt}
],
'temperature': 0.3
}
response = requests.post(
model_config['api_url'],
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
if model_type == 'deepseek':
generated_sql = response.json()['choices'][0]['message']['content'].strip()
else: # OpenAI
generated_sql = response.json()['choices'][0]['message']['content'].strip()
print(f"生成的SQL: {generated_sql}")
return jsonify({'success': True, 'sql': generated_sql})
else:
error_message = f"API调用失败: {response.status_code} - {response.text}"
print(error_message)
return jsonify({
'success': False,
'message': error_message
})
except requests.exceptions.RequestException as e:
error_message = f"API请求错误: {str(e)}"
print(error_message)
return jsonify({
'success': False,
'message': error_message
})
except Exception as e:
error_message = f"SQL生成错误: {str(e)}"
print(error_message)
import traceback
print(traceback.format_exc())
return jsonify({
'success': False,
'message': error_message
})
@main.route('/execute-sql', methods=['POST'])
def execute_sql():
try:
sql = request.json.get('sql')
if not sql or sql.strip().startswith('--'):
return jsonify({
'success': False,
'result': None,
'message': '请先生成有效的 SQL 语句'
})
print(f"执行SQL: {sql}")
result, error_message = db_manager.execute_query(sql)
if result is None:
return jsonify({
'success': False,
'result': None,
'message': error_message or '执行失败，请确保已连接数据库并检查 SQL 语句是否正确'
})
print(f"执行结果: {result}")
return jsonify({
'success': True,
'result': result
})
except Exception as e:
error_message = f"SQL执行错误: {str(e)}"
print(error_message)
import traceback
print(traceback.format_exc())
return jsonify({
'success': False,
'result': None,
'message': error_message
})
@main.route('/get-databases', methods=['GET'])
def get_databases():
databases = db_manager.get_databases()
return jsonify({'databases': databases})
@main.route('/test-connection', methods=['POST'])
def test_connection():
try:
data = request.json
print(f"收到测试连接请求: {data}")
# 创建一个没有指定数据库的连接配置
config = DatabaseConfig(
host=data.get('host'),
user=data.get('user'),
password=data.get('password'),
database='', # 不指定数据库
port=int(data.get('port', 3306))
)
success, error_message = db_manager.connect(config)
if success:
# 获取数据库列表
databases = db_manager.get_databases()
print(f"测试连接成功，获取到数据库列表: {databases}")
return jsonify({
'success': True,
'databases': databases
})
print(f"测试连接失败: {error_message}")
return jsonify({
'success': False,
'message': error_message
})
except Exception as e:
error_message = f"测试连接时发生错误: {str(e)}"
print(error_message)
import traceback
print(traceback.format_exc())
return jsonify({
'success': False,
'message': error_message
}) 到此这篇关于Python调用DeepSeek API实现对本地数据库的AI管理的文章就介绍到这了,更多相关Python DeepSeek数据库管理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
文将介绍8个带有代码的Python项目,这些项目将帮助大家增强编程能力,这些项目涵盖了各种主题和难度级别,助力大家成长为一个Python开发者2024-01-01
在使用 Playwright 进行 Web 自动化时,页面交互是核心操作之一,本文将详细介绍如何使用 Playwright 进行常见的页面交互操作,希望对大家有所帮助2024-10-10
这篇文章主要为大家详细介绍了Python实现ATM系统，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2020-02-02
Cython中提供了一个prange函数，专门用于循环的并行执行。这个 prange的特殊功能是Cython独一无二的，并且prange只能与for循环搭配使用，不能独立存在。本文就将使用 prange 实现 for 循环的并行，感兴趣的可以了解一下2022-08-08
这篇文章主要介绍了python清除指定目录内所有文件中script的方法,涉及Python针对文件、字符串及正则匹配操作的相关技巧,需要的朋友可以参考下2015-06-06
今天小编就为大家分享一篇关于Python编程深度学习绘图库之matplotlib，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧2018-12-12
这篇文章主要为大家详细介绍了Python如何通过三个脚本实现快速整理图片文件,适用于日常数据整理与批量处理场景,感兴趣的小伙伴可以了解下2025-12-12
这篇文章主要介绍了python创建进程fork用法,实例分析了Python使用fork创建进程的使用方法,需要的朋友可以参考下2015-06-06
python shell命令行中import多层目录下的模块操作
这篇文章主要介绍了python shell命令行中import多层目录下的模块操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-03-03
日常办公和数据处理中,我们经常需要将文件从一个目录移动到另一个指定目录,本文就来介绍一下Python实现单个或批量文件移动,具有一定的参考价值,感兴趣的可以了解一下2025-12-12
最新评论