Python調(diào)用DeepSeek?API實(shí)現(xiàn)對本地?cái)?shù)據(jù)庫的AI管理

更新時(shí)間：2025年02月05日 09:56:15 作者：mosquito_lover1

這篇文章主要為大家詳細(xì)介紹了Python如何基于DeepSeek模型實(shí)現(xiàn)對本地?cái)?shù)據(jù)庫的AI管理,文中的示例代碼簡潔易懂,有需要的小伙伴可以跟隨小編一起學(xué)習(xí)一下

場景描述

基于DeepSeek模型，實(shí)現(xiàn)對本地?cái)?shù)據(jù)庫的AI管理。

實(shí)現(xiàn)思路

1、本地python+flask搭建個(gè)WEB，配置數(shù)據(jù)源。

2、通過DeepSeek模型根據(jù)用戶輸入的文字需求，自動(dòng)生成SQL語句。

3、通過SQL執(zhí)行按鈕，實(shí)現(xiàn)對數(shù)據(jù)庫的增刪改查。

前置條件

到DeepSeek官網(wǎng)的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  # 添加這行導(dǎo)入
 
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}")  # 添加調(diào)試信息
        
        # 先創(chuàng)建一個(gè)沒有指定數(shù)據(jù)庫的連接
        config = DatabaseConfig(
            host=data.get('host'),
            user=data.get('user'),
            password=data.get('password'),
            database='',  # 先不指定數(shù)據(jù)庫
            port=int(data.get('port', 3306))
        )
        
        success, error_message = db_manager.connect(config)
        if success:
            # 獲取數(shù)據(jù)庫列表
            databases = db_manager.get_databases()
            print(f"成功獲取數(shù)據(jù)庫列表: {databases}")  # 添加調(diào)試信息
            return jsonify({
                'success': True,
                'databases': databases
            })
        print(f"連接失敗: {error_message}")  # 添加調(diào)試信息
        return jsonify({
            'success': False,
            'message': error_message
        })
    except Exception as e:
        error_message = f"發(fā)生錯(cuò)誤: {str(e)}"
        print(error_message)  # 添加調(diào)試信息
        import traceback
        print(traceback.format_exc())  # 打印完整的錯(cuò)誤堆棧
        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': '請選擇數(shù)據(jù)庫'
            })
            
        print(f"切換到數(shù)據(jù)庫: {database}")
        
        # 使用選擇的數(shù)據(jù)庫重新連接
        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"成功切換到數(shù)據(jù)庫: {database}")
            return jsonify({
                'success': True,
                'message': f'成功切換到數(shù)據(jù)庫: {database}'
            })
        else:
            print(f"切換數(shù)據(jù)庫失敗: {error_message}")
            return jsonify({
                'success': False,
                'message': f'切換數(shù)據(jù)庫失敗: {error_message}'
            })
            
    except Exception as e:
        error_message = f"選擇數(shù)據(jù)庫時(shí)發(fā)生錯(cuò)誤: {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')  # 默認(rèn)值
        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:
            # 使用其他模型的現(xiàn)有邏輯
            model_config = current_app.config['AVAILABLE_MODELS'].get(model_type)
            if not model_config:
                return jsonify({
                    'success': False,
                    'message': '不支持的模型類型'
                })
            
            # 通用提示詞
            prompt = f"""
            作為一個(gè)SQL專家，請根據(jù)以下需求生成合適的SQL語句：
            需求：{user_input}
            
            請只返回SQL語句，不需要其他解釋。
            如果是創(chuàng)建表，請包含合適的字段類型和必要的約束。
            """
 
            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調(diào)用失敗: {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請求錯(cuò)誤: {str(e)}"
                print(error_message)
                return jsonify({
                    'success': False,
                    'message': error_message
                })
 
    except Exception as e:
        error_message = f"SQL生成錯(cuò)誤: {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"執(zhí)行SQL: {sql}")
        result, error_message = db_manager.execute_query(sql)
        
        if result is None:
            return jsonify({
                'success': False,
                'result': None,
                'message': error_message or '執(zhí)行失敗，請確保已連接數(shù)據(jù)庫并檢查 SQL 語句是否正確'
            })
            
        print(f"執(zhí)行結(jié)果: {result}")
        return jsonify({
            'success': True,
            'result': result
        })
        
    except Exception as e:
        error_message = f"SQL執(zhí)行錯(cuò)誤: {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}")
        
        # 創(chuàng)建一個(gè)沒有指定數(shù)據(jù)庫的連接配置
        config = DatabaseConfig(
            host=data.get('host'),
            user=data.get('user'),
            password=data.get('password'),
            database='',  # 不指定數(shù)據(jù)庫
            port=int(data.get('port', 3306))
        )
        
        success, error_message = db_manager.connect(config)
        if success:
            # 獲取數(shù)據(jù)庫列表
            databases = db_manager.get_databases()
            print(f"測試連接成功，獲取到數(shù)據(jù)庫列表: {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"測試連接時(shí)發(fā)生錯(cuò)誤: {str(e)}"
        print(error_message)
        import traceback
        print(traceback.format_exc())
        return jsonify({
            'success': False,
            'message': error_message
        })

到此這篇關(guān)于Python調(diào)用DeepSeek API實(shí)現(xiàn)對本地?cái)?shù)據(jù)庫的AI管理的文章就介紹到這了,更多相關(guān)Python DeepSeek數(shù)據(jù)庫管理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: