Python调用DeepSeek API实现对本地数据库的AI管理！

admin • • Python学习

Python调用DeepSeek API实现对本地数据库的AI管理！

 作者：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

效果展示

202502050954371

核心代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
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管理的文章就介绍到这了。

 

学习资料见知识星球。

以上就是今天要分享的技巧，你学会了吗？若有什么问题，欢迎在下方留言。

快来试试吧，小琥 my21ke007。获取 1000个免费 Excel模板福利​​​​！

更多技巧， www.excelbook.cn

欢迎 加入 零售创新 知识星球，知识星球主要以数据分析、报告分享、数据工具讨论为主；

Excelbook.cn Excel技巧 SQL技巧 Python 学习！

你将获得：

1、价值上万元的专业的PPT报告模板。

2、专业案例分析和解读笔记。

3、实用的Excel、Word、PPT技巧。

4、VIP讨论群，共享资源。

5、优惠的会员商品。

6、一次付费只需129元，即可下载本站文章涉及的文件和软件。

文章版权声明 1、本网站名称：Excelbook
2、本站永久网址：http://www.excelbook.cn
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长王小琥进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报。
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END
分享
二维码
< <上一篇
下一篇>>