Dify RAGFlow：泵类设备预测维护系统案例分享

上篇文章介绍到的 dify+RAGFLow 的协同使用文章里，提到了一个泵类设备预测性维护智能系统。后来陆续有人私信咨询实施细节，这篇做个统一的介绍。

Dify+RAGFlow:1+1＞2的混合架构，详细教程+实施案例

项目定位是，利用 Dify 的工作流编排能力和 RAGFlow 的知识库组件，结合模拟的设备传感器数据 (IoT) 和企业资源数据 (CMMS, MES, ERP)，构建一个针对离心式冷却液泵的预测性维护系统原型。

注：为遵守保密协议，本文及相关代码在原项目基础上使用了模拟数据和接口进行演示。所展示的工作流设计、节点交互等与实际项目逻辑相似，仅供各位参考。

以下，enjoy:

1

项目背景

作者早年从事金融信贷领域时，在长珠三角走访过大大小小几百家工厂，其中以机械加工、注塑行业为主，但不管什么细分行业，设备的稳定运行对企业来说都是日常经营的头等大事。尤其是像 CNC 加工中心使用的冷却液泵这类关键辅助设备，意外停机会导致生产中断，直接影响订单排查计划。

传统的定期维护 (Preventive Maintenance, PM) 往往过于保守或未能及时发现潜在问题，而事后维修 (Corrective Maintenance, CM) 成本高昂。这篇要介绍的预测性维护 (Predictive Maintenance, PdM) 主要目的是通过监测离心泵设备状态，在故障实际发生前安排维护，最大限度地减少停机时间、降低维护成本、提高设备利用率。

图片源于网络搜索，侵删

泵类型:离心式冷却液泵 (Centrifugal Coolant Pump)
泵型号 :CoolFlow Pro 5000
设备 IDUMP-CNC-001
使用场景:安装在一台中型 CNC 立式加工中心 (Machine ID: VMC-03) 上，用于循环供给水基切削液，冷却刀具和工件。该机床主要加工钢材和铝合金零件，实行两班倒工作制 (16小时/天)。泵已运行约 18 个月。
关键监测参数:振动 (mm/s RMS), 温度 (°C), 出口压力 (Bar), 流量 (L/min)。

2

数据构成

参照真实的业务场景，我构建了一套模拟 API (mock_api.py) 和相应的 JSON 数据文件，代表不同系统的数据源：

1、IoT 数据 (`IOT.json` & `/api/pump/status`):
  *  包含泵的实时传感器读数，如时间戳 (`timestamp`)、轴向/径向振动 (`vibration_axial`/`radial`)、电机/轴承温度 (`temperature_motor`/`bearing`)、出口压力 (`pressure_outlet`)、流量 (`flow_rate`)。  *  模拟了数据随时间变化，包含正常和逐渐异常的状态。
2、MES 数据 (`MES.json` & `/api/mes/pump/operation`):  *  提供泵的运行日志，关联的设备 (`associated_machineId`)，运行小时数 (`operating_hours`)，处理的工单 (`jobs_processed`) 等。  *  有助于了解泵的工作负载和上下文。
3、CMMS 数据 (`CMMS.json` & `/api/cmms/pump/history`):  *  包含历史维护记录 (`maintenance_records`)：工单号、日期、类型、描述、更换部件、停机时间等。  *  包含相关故障信息 (`related_failures_info`)：其他泵（或本机历史）的故障案例，包括症状、故障模式、根本原因、纠正措施。这是重要的知识来源。
4、ERP 数据 (`ERP.json` & `/api/erp/spare_parts`):  *  提供备件信息：零件 ID (`partId`)、名称 (`name`)、描述、适用泵型号、供应商、库存水平 (`stock_level`)、单价 (`unit_price`)、采购提前期 (`lead_time_days`)。  *  用于生成维护建议时考虑备件可用性。
模拟维护完成后，实际执行情况的反馈数据，用于评估预测准确性和闭环优化。

3

工作流整体思路

由 RAGFlow 负责处理知识库，工作流基于 Dify 构建，核心思路是：

状态监测 -> 异常判断 -> 深度分析与建议 -> 报告生成。

获取状态:通过 GET_IOT_DATA 节点调用模拟 API 获取指定 pumpId 的最新传感器数据。

解析数据:使用 CODE 节点解析返回的 JSON 字符串，提取关键指标，并判断是否需要维护 (needs_maintenance 标志)。

条件分流:IF/ELSE 节点根据 needs_maintenance 标志决定流程走向。

ELSE(正常):流向ANSWER节点，输出简单的状态正常信息。

IF(异常):启动预测性维护流程。

信息整合 (IF 分支):

调用GET_CMMS_DATA和GET_MES_DATA获取历史维护和运行数据。

调用KNOWLEDGE_RETRIEVAL节点，结合当前异常指标查询知识库(离心泵设备手册pdf、历史维修案例excel)。

故障预测 (IF 分支):调用 FAULT_PREDICTION(LLM 节点)，将整合后的信息（实时数据、历史数据、知识库信息）提供给 LLM，要求其分析可能的故障模式并预测需要更换的关键备件 ID。

备件 ID 提取 (IF 分支):使用 EXTRACT_PART_ID (CODE 节点) 从 FAULT_PREDICTION 的文本输出中解析出 required_partId。

ERP 查询 (IF 分支):调用 GET_ERP_DATA (HTTP 节点)，使用 extracted_part_id 作为参数查询具体备件信息 。

ERP 数据解析 (IF 分支):使用 PARSE_ERP_DATA(CODE 节点) 解析 GET_ERP_DATA 返回的 JSON 字符串为列表。

生成维护建议 (IF 分支):调用 SUGGESTIONS(LLM 节点)，将故障预测结果和解析后的 ERP 备件信息列表提供给 LLM，要求生成详细的维护步骤、备件清单和时间建议。

最终输出:通过 ANSWER 2 节点展示维护建议或最终报告。

4

关键节点介绍

 **START:** 定义工作流入口参数 (`pumpId`)。 **GET\_... (HTTP Request):** 与模拟 API 交互，获取各类数据。 **CODE (解析 IoT):** 解析 IoT JSON，提取关键指标，**计算 `needs_maintenance` 标志**。 **IF/ELSE:** 根据 `needs_maintenance` 标志进行流程分支。 **KNOWLEDGE_RETRIEVAL:** (可选) 连接外部知识库进行 RAG 查询。 **FAULT_PREDICTION (LLM):** 基于多源信息进行故障诊断与预测。 **EXTRACT_PART_ID (CODE):** 从 LLM 输出中解析结构化信息 (Part ID)。 **PARSE_ERP_DATA (CODE):** 解析 ERP 返回的 JSON 列表字符串。 **SUGGESTIONS (LLM):** 结合故障预测和备件信息生成维护建议。 **ANSWER / ANSWER 2:** 输出最终结果。

5

主要报错与解决过程参考

基于我前期在上手熟悉 Dify 工作流搭建中踩过的坑，结合这个项目做个对照说明，各位可以大致做个参考：

1. HTTP 请求：URL 与参数混淆  *  **节点:** `GET_IOT_DATA`  *  **报错:** `Invalid non-printable ASCII character...`  *  **原因:** 同时在 URL 字段和 PARAMS 字段中定义了查询参数。  *  **解决:** URL 只写基础路径，参数在 PARAMS 中定义 (推荐: `{{start.pumpId}}`)。
2. HTTP 请求：网络访问 & CORS  *  **节点:** 所有 HTTP 请求节点  *  **报错:** Dify 404，但 API 日志显示 200 OK。  *  **原因:** API 服务监听地址限制；缺少 CORS 配置。  *  **解决:**    *  API (`mock_api.py`): 启动时 `host='0.0.0.0'`。    *  API (`mock_api.py`): 安装 `flask-cors` 并启用 `CORS(app)`。
3. 数据处理：JSON 字符串解析  *  **节点:** `CODE` (解析 IoT), `IF/ELSE`, `ANSWER`  *  **报错:** `IF/ELSE` 条件不生效；`ANSWER` 直接输出占位符。  *  **原因:** HTTP 节点返回的 `body` 是 JSON **字符串**，而非结构化对象。  *  **解决:** 在 HTTP 节点后加 `CODE` 节点，用 `json.loads()` 解析 `body` 字符串，输出解析后的对象/列表供后续节点使用。
4. IF/ELSE 节点：数值比较类型问题  *  **节点:** `IF/ELSE`  *  **报错:** 数值比较条件 (`> 6.0`) 不生效。  *  **原因:** `IF/ELSE` 可能将变量视为字符串进行比较。  *  **解决:** 将比较逻辑移入 `CODE` 节点，输出简单标志位 (如 `"1"`/`"0"` 字符串)，`IF/ELSE` 只判断此标志位。
5. CODE 节点：Python 语法 & 类型错误  *  **节点:** 所有 `CODE` 节点  *  **报错:** `IndentationError`, `NameError`, `Output variable 'xxx' must be a [Type]`。  *  **原因:** 缩进错误；变量未定义；Dify 输出类型配置与代码返回不符。  *  **解决:** 严格检查 Python 缩进；确保 `return` 前所有变量已定义赋值；确保 Dify 中输出变量的**名称和类型**与代码 `return` **完全匹配** (e.g., Part ID 是 String)。
6. TEMPLATE 节点：Jinja2 语法 & 数据错误  *  **节点:** `TEMPLATE`  *  **报错:** `jinja2.exceptions.TemplateSyntaxError`。  *  **原因:** 括号错误 (`{{{` vs `{{`, `}} }`); 控制结构括号错误 (`{{%` vs `{%`); 变量名不匹配；在未解析数据上访问属性。  *  **解决:** 修正 Jinja2 语法；确保模板变量名与输入变量名**完全一致**；对列表/字典数据，**先用 CODE 解析**再传入模板，并使用 `{% for %}` 访问循环变量属性 (`{{ part.name }}`)。

6

从模拟到生产切换注意事项

看完上述内容，如果你计划将基于 mock_api.py 测试成功的工作流切换到对接真实的生产环境，需要注意替换数据源接入点，并适配数据格式。

主要步骤如下：

1.获取真实API信息:**获取真实的IoT、CMMS、MES、ERP系统的API详细信息：***URL端点(Endpoint):**每个系统提供数据的具体网址。***请求方法(Method):**GET,POST,PUT,DELETE等。***认证方式(Authentication):**如何验证Dify的访问权限？可能是APIKey(放在Header或QueryParam)、OAuth2.0Token(放在Header)、用户名密码等。***请求参数/体(Parameters/Body):**如何指定要查询的设备ID、时间范围、零件ID等？是在URL参数中，还是在POST请求的Body中？Body的格式是JSON、XML还是其他？***响应格式(ResponseFormat):**真实系统返回的数据结构是怎样的？与模拟的JSON结构是否一致？2.修改DifyHTTPRequest节点:*定位工作流中所有调用模拟API的HTTPRequest节点(`GET_IOT_DATA`,`GET_CMMS_DATA`,`GET_MES_DATA`,`GET_ERP_DATA`)。***更新URL:**将节点的URL字段替换为真实API的端点。***调整方法:**根据真实API要求修改请求方法(GET/POST等)。***配置认证:**在节点的`Headers`或`Params`部分添加必要的认证信息(如`Authorization:Bearer<token>`,`X-API-Key:<key>`)。***适配参数/体:**根据真实API要求修改`Params`或`Body`的内容和格式。3.适配数据解析(关键步骤❗️):*真实API返回的数据结构**极有可能**与模拟JSON不同。*因此，所有紧跟在HTTPRequest节点之后的`CODE`节点（如解析IoT的`CODE`节点,`PARSE_ERP_DATA`节点，以及可能需要为CMMS/MES新增的解析节点）**必须进行修改**。*需要根据真实API返回的JSON(或其他格式)结构，调整Python代码中`json.loads()`之后访问数据的方式（字典键名、列表索引、嵌套结构等），以确保存储到Dify输出变量中的是后续节点期望的数据。***这是最容易出错的环节，需要仔细比对真实API的响应并耐心调试**`CODE`**节点的Python代码。4.端到端测试:**修改完成后，使用真实的设备ID或测试ID进行完整的端到端测试，验证：*是否能成功从所有真实系统中获取数据。*数据解析是否正确。*LLM节点接收到的上下文是否符合预期。*整个流程是否能按预期运行。

网络连通性:确保运行 Dify 的环境能够访问到企业内网的真实系统 API 地址。

权限管理:确保 Dify 使用的 API Key 或凭证具有访问所需数据的正确权限。

数据格式差异:重点关注真实 API 与模拟 API 在数据结构上的差异，并适配解析逻辑。

性能与频率:真实 API 可能有调用频率限制或响应时间较长，需要考虑工作流的性能和可能的超时设置。

错误处理:针对真实 API 可能出现的各种错误（网络错误、认证失败、无效参数、系统内部错误等），在 Dify 工作流中增加更健壮的错误处理逻辑。

7

后续拓展方向

增加报告模板(TEMPLATE):引入并调试TEMPLATE节点，利用 Jinja2 生成结构更清晰、格式更专业的 HTML、Markdown、PDF 报告 (需要额外工具集成)。

定时触发与批量处理:增加定时任务触发器 (如 Dify 的计划任务或外部调度系统调用 API)，定期检查设备状态，而不是仅通过手动输入pumpId触发。

集成到业务系统:

告警通知:在检测到异常时，通过HTTP请求节点调用企业微信、钉钉、邮件等的API发送告警通知。

工单系统集成:将`SUGGESTIONS`生成的维护建议通过API推送到CMMS系统，自动创建维护工单。

API 暴露:将整个 Dify 工作流封装为 API，供其他业务系统（如设备监控大屏、生产管理系统）调用。

闭环反馈机制:实现/api/cmms/report/update接口，接收维护完成后的实际结果。设计新的 Dify 工作流或节点来处理这些反馈，分析预测准确性，并将分析结果存入数据库或知识库，用于持续优化模型或 Prompt。

更复杂的故障预测模型:FAULT_PREDICTION 可以替换为更专业的机器学习模型服务接口，或者使用更强大的 LLM 并优化 Prompt 以进行更精确的故障定位和剩余寿命预测 (RUL)。

知识库增强:持续丰富 RAGFlow (MinerU、 Mistral OCR等工具的测评后续会专门写一篇) 中的内容，加入更多设备类型、故障案例和解决方案。

用户交互优化:利用 Dify 的 Agent 或 Web App 能力，创建更友好的用户界面，允许用户查询特定设备状态、历史记录或手动触发分析。