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数据库管理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
今天小编就为大家分享一篇TensorFlow梯度求解tf.gradients实例，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-02-02
这篇文章主要为大家详细介绍了python自动计算图像数据集的RGB均值，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-06-06
今天小编就为大家分享一篇Python学习_几种存取xls/xlsx文件的方法总结，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-05-05
这篇文章主要介绍了python用Configobj模块读取配置文件，帮助大家更好的利用python处理文件，感兴趣的朋友可以了解下2020-09-09
这篇文章主要介绍了pandas 按日期范围筛选数据的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2021-02-02
今天小编就为大家分享一篇解决Shell执行python文件,传参空格引起的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-10-10
这篇文章主要介绍了python实现微秒级等待问题(windows),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-06-06
基于keras中训练数据的几种方式对比(fit和fit_generator)
这篇文章主要介绍了keras中训练数据的几种方式对比(fit和fit_generator)，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-05-05
本文主要介绍了python爬虫字体加密的解决，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2023-03-03
PyQt5是一个Python绑定库,用于Qt C++ GUI框架,它允许开发者使用Python语言创建跨平台的应用程序,并利用丰富的Qt图形用户界面功能,本文介绍了python中PyQt5窗口界面的各种交互逻辑实现,需要的朋友可以参考下2024-07-07
最新评论