DeepSeek+dify 工作流应用,自然语言查询数据库信息并展示

引言

在写连接外部数据库查询的时候，我就想普通的业务人员没有编程背景，不懂数据库，可他们想查询统计一些东西怎么办？只能找对应的开发人员写sql了，那我是不是可以用工作流解决这个问题。

简单的梳理了下流程。

代码示例

我先把在 xxx里写的python脚本改下，由关键词检索改成sql检索。

from flask import Flask, request, jsonify

import pymysql



app = Flask(__name__)



# 数据库配置

DATABASE_CONFIG = {

    'host': '', # 自己的数据库地址

    'user': '', # 自己数据库的账户

    'password': '', #自己数据库的密码

    'db': 'demo', # 自己数据库的库名

    'charset': 'utf8mb4',

    'cursorclass': pymysql.cursors.DictCursor

}



@app.route('/query', methods=['POST'])

def query_database():

    print("接收到请求")

    # 获取查询sql

    querySql = request.json.get('querySql')

    print("querySql为:"+querySql)

    # 去除 Markdown 标识和不必要的字符

    if querySql:

        # 去除开头的 ````sql`

        querySql = querySql.replace('```sql\n', '')

        

        # 去除结尾的 ``<EOF>``

        querySql = querySql.replace('\n<<EOF>>', '')

        

        # 去除多余的空格和换行符

        querySql = querySql.strip()

        # 将换行符替换为空格（如果需要保持多行，可以省略这一步）

        querySql = querySql.replace('\n', ' ')

        querySql = querySql.replace('```', ' ')

        # 去除多余的空格

        querySql = ' '.join(querySql.split())

        

    ifnot querySql:

        return jsonify({"error": "querySql is required"}), 400

    

    try:

        # 建立数据库连接

        connection = pymysql.connect(**DATABASE_CONFIG)

        with connection.cursor() as cursor:

            # 执行查询

            cursor.execute(querySql)

            result = cursor.fetchall()

        

        connection.commit()

        connection.close()



        ifnot result:

            return"未查询到有效数据", 400

        

        # 生成 Markdown 表格

        markdown_table = generate_markdown_table(result)

        

        return markdown_table, 200

    

    except Exception as e:

        return str(e), 500



def generate_markdown_table(results):

    """ 生成 Markdown 表格 """

    ifnot results:

        return""

    # 获取列名

    columns = results[0].keys()

    

    # 表头

    table_md = "| " + " | ".join([col for col in columns]) + " |\n"

    # 分隔线

    table_md += "| " + " --- |" * len(columns) + "\n"

    

    # 表格内容

    for row in results:

        table_md += "| " + " | ".join([str(cell) for cell in row.values()]) + " |\n"

    

    return table_md



if __name__ == '__main__':

    app.run(host='10.1.0.65', port=8000)