Dify 数据分析工作流搭建

本节目标

1、理解数据分析工作流的整体架构

2、掌握数据库结构自动获取方法

3、掌握基于大模型的 SQL 自动生成能力

一完整流程图

数据分析工作流流程图如下所示：

1、获取数据库描述

2、SQL语句生成并查询回复

二获取数据库信息模块工作流

1、数据库查询节点

（1）设置数据库查询节点

（2）设置数据库查询节点

（3）配置节点属性

数据库类型：MySQL

数据库地址：dify安装在docker中，我们填写：host.docker.internal 获取宿主机的地址。

也可以直接填写宿主机的ip地址：192.168.xx.xx

端口：3306

用户名：数据库的用户名

密码：数据库的密码

库名：数据库名称，这里填写：sales_analysis_db

（4）设置SQL语句

SQL查询语句：

SELECT
    TABLE_NAME,
    TABLE_COMMENT
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'sales_analysis_db';

输出格式：有两个可选，markdown格式或者json格式。

因为我们要获取数据库名称，这里选择json格式。

（5）试运行观察结果

试运行发现得到结果如下：

{
  "text": "",
  "files": [],
  "json": [
    {
      "records": [
        {
          "TABLE_COMMENT": "订单明细表，存储每个订单中的商品信息",
          "TABLE_NAME": "order_item"
        },
        {
          "TABLE_COMMENT": "订单表，存储每一笔订单",
          "TABLE_NAME": "orders"
        },
        {
          "TABLE_COMMENT": "商品表，存储商品信息",
          "TABLE_NAME": "product"
        },
        {
          "TABLE_COMMENT": "用户表，存储用户基础信息",
          "TABLE_NAME": "user"
        }
      ]
    }
  ]
}

2、代码节点-获取表名字

（1）设置代码执行节点

（2）配置代码执行节点参数：

输入变量：上一步查询所有数据表的输出变量

python代码：可以通过以下代码获取表名以及表名对应的说明

def main(json):
    json = json[0]
    records = json.get("records", [])
    table_name_list = []
    table_name_comment_dict = {}
    for record in records:
        table_name = record.get("TABLE_NAME", "")
        table_comment = record.get("TABLE_COMMENT", "")
        table_name_list.append(table_name)
        table_name_comment_dict[table_name] = table_comment
    return {
        "table_name_list": table_name_list,
        "table_name_comment_dict": table_name_comment_dict,
    }

3、迭代节点-循环判断

（1）设置迭代节点

如图所示设置迭代节点

输入的为一个列表，就是上一步输出的list

（2）数据库查询节点-查询表结构

我们有了表结构，我们才可以根据表结构生成SQL语句。

继续设置数据库查询节点，配置和之前一样。

SQL查询语句：

SELECT
    COLUMN_NAME,
    COLUMN_TYPE,
    COLUMN_COMMENT
FROM information_schema.COLUMNS
WHERE TABLE_NAME = ''
  AND TABLE_SCHEMA = 'sales_analysis_db';

（3）数据库查询节点-查询表例子

我们给大模型一些示例数据，大模型才可以根据示例数据生成更精确的SQL语句。

（4）代码节点-生成表描述

输入变量：

1、item：表名。

2、column_text：SQL查询表结构的结果。

3、example_text：SQL查询表例子的结果。

4、table_name_comment_dict：获取表名对应说明的dict。

代码配置如下：

def main(item:str, column_text: str, example_text: str, table_name_comment_dict):
    comment = table_name_comment_dict.get(item, "")
    result_text = f"""
    数据表名：{item}
    数据表描述：{comment}
    数据表字段及描述：
    {column_text}
    数据表例子(示例数据)：
    {example_text}
    """
    return {
        "result": result_text
    }

输出变量：result

4、代码节点-整合多个表描述

（1）配置节点

输入变量：table_str_list

代码如下：

def main(table_str_list):
    result_str = ""
    result_str += "所有的表结构及描述如下:\n"
    for item in table_str_list:
        result_str += item
        result_str += "\n\n"
    return {
        "result": result_str
    }

输出变量：result

（2）试运行观察结果

三 SQL语句生成并回复

1、输出SQL语句

系统提示词：

你是一个专业的SQL生成助手，请根据用户的自然语言问题，生成【MySQL查询SQL语句】。

====================
【数据库结构信息】
====================

{{#1775219864143.result#}}

====================
【任务要求】
====================

1. 你只能生成【SELECT查询语句】，严禁生成：
- INSERT
- UPDATE
- DELETE
- DROP
- ALTER
- SHOW
- DESC

2. 字段使用规则：
- 只能使用已提供表中的字段
- 所有字段必须带表别名
- 多表查询必须优先使用 JOIN
- 禁止臆造不存在的字段和表

3. 查询规范：
- 默认加 LIMIT 10，除非用户明确要求全部数据
- 涉及“最多 / 最少 / TOP / 排名”时，必须使用 GROUP BY + ORDER BY + LIMIT
- 涉及金额统计时，优先使用 SUM / AVG / MAX / MIN
- 涉及时间时，优先使用 order_time 或 register_time
- 排序默认优先使用 DESC

4. 输出必须严格遵循 JSON 格式：
- 只能输出一个合法 JSON 对象
- 不要输出解释
- 不要输出 markdown
- 不要输出多余文字
- 不要使用 ```json 或 ```sql 包裹

====================
【输出JSON格式】
====================

{
"sql": "生成的SQL语句",
"question": "用户原始问题",
"tables": ["本次SQL涉及的表名1", "本次SQL涉及的表名2"],
"summary": "一句话说明这条SQL的查询意图"
}

====================
【输出要求补充】
====================

1. sql 字段中只放 SQL 语句本身
2. question 字段中保留用户原始问题
3. tables 字段中列出实际使用到的表名
4. summary 字段用一句简洁中文说明SQL用途
5. 输出内容必须是标准 JSON，可被程序直接解析
6. 若用户问题无法映射到当前表结构，也必须输出标准 JSON，格式如下：

{
"sql": "",
"question": "用户原始问题",
"tables": [],
"summary": "无法根据当前表结构生成SQL"
}

====================
【示例】
====================

用户问题：
每个城市有多少用户？

输出：
{
"sql": "SELECT u.city AS city, COUNT(*) AS user_count FROM user u GROUP BY u.city ORDER BY user_count DESC LIMIT 10;",
"question": "每个城市有多少用户？",
"tables": ["user"],
"summary": "统计每个城市的用户数量并按数量倒序排序"
}

用户提示词：

用户问题：{{#sys.query#}}

记忆：

可以配置记忆，使其涵盖记忆效果。

2、Json解析对象

（1）安装工具

安装json解析工具，如下图所示

（2）对生成的SQL进行JSON解析

解析对象：sql

我们从而获取到sql。

3、查询数据库

设置查询数据库模块。

SQL查询语句：这里SQL查询语句为上一步JSON解析的。

输出格式：这里设置为markdown

4、大模型LLM-根据查询结果回答