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

场景描述

基于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数据库管理内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！