调用DeepSeek API实现对当地数据库的AI管理

场景形貌：基于DeepSeek模型，实现对当地数据库的AI管理。
实现思路：
1、当地python+flask搭建个WEB，设置数据源。
2、通过DeepSeek模型根据用户输入的文字需求，主动生成SQL语句。
3、通过SQL实行按钮，实现对数据库的增删改查。
前置条件：到DeepSeek官网的API开放平台注册，完成以下设置：

1. DEEPSEEK_API_KEY=your-deepseek-api-key
2. DEEPSEEK_API_URL=https://api.deepseek.com/v1/chat

复制代码

结果展示：

核心代码：

1. from flask import Blueprint, render_template, request, jsonify, current_app
2. from .database import DatabaseManager
3. from .config import DatabaseConfig
4. import requests
5. from .sql_generator import SQLGenerator  # 添加这行导入
7. main = Blueprint('main', __name__)
8. db_manager = DatabaseManager()
10. @main.route('/')
11. def index():
12.     return render_template('index.html')
14. @main.route('/connect', methods=['POST'])
15. def connect_database():
16.     try:
17.         data = request.json
18.         print(f"收到连接请求: {data}")  # 添加调试信息
19.         
20.         # 先创建一个没有指定数据库的连接
21.         config = DatabaseConfig(
22.             host=data.get('host'),
23.             user=data.get('user'),
24.             password=data.get('password'),
25.             database='',  # 先不指定数据库
26.             port=int(data.get('port', 3306))
27.         )
28.         
29.         success, error_message = db_manager.connect(config)
30.         if success:
31.             # 获取数据库列表
32.             databases = db_manager.get_databases()
33.             print(f"成功获取数据库列表: {databases}")  # 添加调试信息
34.             return jsonify({
35.                 'success': True,
36.                 'databases': databases
37.             })
38.         print(f"连接失败: {error_message}")  # 添加调试信息
39.         return jsonify({
40.             'success': False,
41.             'message': error_message
42.         })
43.     except Exception as e:
44.         error_message = f"发生错误: {str(e)}"
45.         print(error_message)  # 添加调试信息
46.         import traceback
47.         print(traceback.format_exc())  # 打印完整的错误堆栈
48.         return jsonify({
49.             'success': False,
50.             'message': error_message
51.         })
53. @main.route('/select-database', methods=['POST'])
54. def select_database():
55.     try:
56.         data = request.json
57.         database = data.get('database')
58.         if not database:
59.             return jsonify({
60.                 'success': False,
61.                 'message': '请选择数据库'
62.             })
63.             
64.         print(f"切换到数据库: {database}")
65.         
66.         # 使用选择的数据库重新连接
67.         config = DatabaseConfig(
68.             host=db_manager.host,
69.             user=db_manager.user,
70.             password=db_manager.password,
71.             database=database,
72.             port=db_manager.port
73.         )
74.         
75.         success, error_message = db_manager.connect(config)
76.         if success:
77.             print(f"成功切换到数据库: {database}")
78.             return jsonify({
79.                 'success': True,
80.                 'message': f'成功切换到数据库: {database}'
81.             })
82.         else:
83.             print(f"切换数据库失败: {error_message}")
84.             return jsonify({
85.                 'success': False,
86.                 'message': f'切换数据库失败: {error_message}'
87.             })
88.             
89.     except Exception as e:
90.         error_message = f"选择数据库时发生错误: {str(e)}"
91.         print(error_message)
92.         import traceback
93.         print(traceback.format_exc())
94.         return jsonify({
95.             'success': False,
96.             'message': error_message
97.         })
99. @main.route('/generate-sql', methods=['POST'])
100. def generate_sql():
101.     try:
102.         user_input = request.json.get('input')
103.         model_type = request.json.get('model')  # 默认值
104.         print(f"收到SQL生成请求，用户输入: {user_input}，选择模型: {model_type}")
106.         if model_type == '':
107.             # 使用本地 SQL 生成器
108.             generated_sql = SQLGenerator.generate_sql(user_input)
109.             return jsonify({
110.                 'success': True,
111.                 'sql': generated_sql
112.             })
113.         else:
114.             # 使用其他模型的现有逻辑
115.             model_config = current_app.config['AVAILABLE_MODELS'].get(model_type)
116.             if not model_config:
117.                 return jsonify({
118.                     'success': False,
119.                     'message': '不支持的模型类型'
120.                 })
121.             
122.             # 通用提示词
123.             prompt = f"""
124.             作为一个SQL专家，请根据以下需求生成合适的SQL语句：
125.             需求：{user_input}
126.             
127.             请只返回SQL语句，不需要其他解释。
128.             如果是创建表，请包含合适的字段类型和必要的约束。
129.             """
131.             try:
132.                 headers = {
133.                     'Authorization': f'Bearer {model_config["api_key"]}',
134.                     'Content-Type': 'application/json'
135.                 }
137.                 if model_type == 'deepseek':
138.                     payload = {
139.                         'model': model_config['model_name'],
140.                         'messages': [
141.                             {'role': 'user', 'content': prompt}
142.                         ],
143.                         'temperature': 0.3
144.                     }
145.                 else:  # OpenAI
146.                     payload = {
147.                         'model': model_config['model_name'],
148.                         'messages': [
149.                             {'role': 'system', 'content': 'You are a SQL expert. Only return SQL statements without any explanation.'},
150.                             {'role': 'user', 'content': prompt}
151.                         ],
152.                         'temperature': 0.3
153.                     }
155.                 response = requests.post(
156.                     model_config['api_url'],
157.                     headers=headers,
158.                     json=payload,
159.                     timeout=30
160.                 )
162.                 if response.status_code == 200:
163.                     if model_type == 'deepseek':
164.                         generated_sql = response.json()['choices'][0]['message']['content'].strip()
165.                     else:  # OpenAI
166.                         generated_sql = response.json()['choices'][0]['message']['content'].strip()
167.                     
168.                     print(f"生成的SQL: {generated_sql}")
169.                     return jsonify({'success': True, 'sql': generated_sql})
170.                 else:
171.                     error_message = f"API调用失败: {response.status_code} - {response.text}"
172.                     print(error_message)
173.                     return jsonify({
174.                         'success': False,
175.                         'message': error_message
176.                     })
178.             except requests.exceptions.RequestException as e:
179.                 error_message = f"API请求错误: {str(e)}"
180.                 print(error_message)
181.                 return jsonify({
182.                     'success': False,
183.                     'message': error_message
184.                 })
186.     except Exception as e:
187.         error_message = f"SQL生成错误: {str(e)}"
188.         print(error_message)
189.         import traceback
190.         print(traceback.format_exc())
191.         return jsonify({
192.             'success': False,
193.             'message': error_message
194.         })
196. @main.route('/execute-sql', methods=['POST'])
197. def execute_sql():
198.     try:
199.         sql = request.json.get('sql')
200.         if not sql or sql.strip().startswith('--'):
201.             return jsonify({
202.                 'success': False,
203.                 'result': None,
204.                 'message': '请先生成有效的 SQL 语句'
205.             })
207.         print(f"执行SQL: {sql}")
208.         result, error_message = db_manager.execute_query(sql)
209.         
210.         if result is None:
211.             return jsonify({
212.                 'success': False,
213.                 'result': None,
214.                 'message': error_message or '执行失败，请确保已连接数据库并检查 SQL 语句是否正确'
215.             })
216.             
217.         print(f"执行结果: {result}")
218.         return jsonify({
219.             'success': True,
220.             'result': result
221.         })
222.         
223.     except Exception as e:
224.         error_message = f"SQL执行错误: {str(e)}"
225.         print(error_message)
226.         import traceback
227.         print(traceback.format_exc())
228.         return jsonify({
229.             'success': False,
230.             'result': None,
231.             'message': error_message
232.         })
234. @main.route('/get-databases', methods=['GET'])
235. def get_databases():
236.     databases = db_manager.get_databases()
237.     return jsonify({'databases': databases})
239. @main.route('/test-connection', methods=['POST'])
240. def test_connection():
241.     try:
242.         data = request.json
243.         print(f"收到测试连接请求: {data}")
244.         
245.         # 创建一个没有指定数据库的连接配置
246.         config = DatabaseConfig(
247.             host=data.get('host'),
248.             user=data.get('user'),
249.             password=data.get('password'),
250.             database='',  # 不指定数据库
251.             port=int(data.get('port', 3306))
252.         )
253.         
254.         success, error_message = db_manager.connect(config)
255.         if success:
256.             # 获取数据库列表
257.             databases = db_manager.get_databases()
258.             print(f"测试连接成功，获取到数据库列表: {databases}")
259.             return jsonify({
260.                 'success': True,
261.                 'databases': databases
262.             })
263.         print(f"测试连接失败: {error_message}")
264.         return jsonify({
265.             'success': False,
266.             'message': error_message
267.         })
268.     except Exception as e:
269.         error_message = f"测试连接时发生错误: {str(e)}"
270.         print(error_message)
271.         import traceback
272.         print(traceback.format_exc())
273.         return jsonify({
274.             'success': False,
275.             'message': error_message
276.         })

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。