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

引言

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

简单的梳理了下流程。

代码示例

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

if name == ‘main‘:
app.run(host=’10.1.0.65’, port=8000)

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('```sqln', '')

# 去除结尾的 `<EOF>`
        querySql = querySql.replace('n<>', '')

# 去除多余的空格和换行符
        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)