借助Dify连接MySQL数据库

借助Dify连接MySQL数据库

1、在本地搭建数据库访问服务并通过ngrok暴露至公网

在本地搭建能够访问数据库的服务，并且利用ngrok将该服务暴露到公网环境中。

#sql_tools.py

from flask import Flask, request, jsonify
import mysql.connector

# 数据库连接配置
config = {
    'user': 'your_username',
    'password': 'your_password',
    'host': 'localhost',
    'database': 'your_database',
    'raise_on_warnings': True
}

# 初始化Flask应用
app = Flask(__name__)

# 连接数据库
def connect_to_database():
    try:
        conn = mysql.connector.connect(**config)
        print("已连接到MySQL数据库")
        return conn
    except mysql.connector.Error as err:
        print(f"错误：{err}")
        return None

# 执行SQL查询
def execute_query(conn, sql):
    cursor = conn.cursor()
    try:
        cursor.execute(sql)
        if sql.strip().lower().startswith("select"):
            # 如果是查询操作，返回结果
            result = cursor.fetchall()
            return result
        else:
            # 如果是插入、更新、删除操作，提交事务并返回受影响的行数
            conn.commit()
            return cursor.rowcount
    except mysql.connector.Error as err:
        print(f"执行SQL时出错：{err}")
        return None
    finally:
        cursor.close()

# HTTP接口：执行SQL
@app.route('/execute', methods=['POST'])
def execute_sql():
    # 获取请求中的SQL语句
    data = request.json
    if not data or 'sql' not in data:
        return jsonify({"error": "需要SQL语句"}), 400

    sql = data['sql']
    conn = connect_to_database()
    if not conn:
        return jsonify({"error": "无法连接到数据库"}), 500

    # 执行SQL
    result = execute_query(conn, sql)
    conn.close()

    if result is None:
        return jsonify({"error": "执行SQL失败"}), 500

    # 返回结果
    return jsonify({"result": result})

# 启动Flask应用
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3000)

2、创建知识库并导入表结构描述

创建一个知识存储的库，然后把表格结构的相关描述导入进去。

3、创建数据库访问工作流程

构建针对数据库访问的工作流程。

import requests
def main(sql: str) -> dict:
    # 定义API的URL
    url = "https://xxx.ngrok-free.app/execute"

    # 构造请求体
    payload = {
        "sql": sql
    }

    # 发送POST请求
    try:
        response = requests.post(url, json=payload)

        # 检查响应状态码
        if response.status_code == 200:
            # 解析响应数据
            result = response.json()
            return {
                "result": f"{result}"
            }
        else:
            return {
                "result": f"请求失败，状态码：{response.status_code},{response.json()}"
            }
    except requests.exceptions.RequestException as e:
        return {
            "result": f"请求异常：{e}"
        }

4、创建数据库智能体

构建数据库相关的智能体。

方法二：直接在执行代码中访问mysql

导入pymysql模块，定义执行SQL的函数以及主函数来调用。同时需要进行配置：
1、在dify/docker-legacy/volumes/sandbox/dependencies/python-requirements.txt中添加pymysql这个依赖项
2、在dify/docker/ssrf_proxy/squid.conf.template中添加对数据库IP及端口的访问权限。

参考：如果不是本地网络环境，可以使用devnet。