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
設備 ID: PUMP-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 的訪問權限？可能是 API Key (放在 Header 或 Query Param)、OAuth 2.0 Token (放在 Header)、用戶名密碼等。
* **請求參數/體 (Parameters/Body):** 如何指定要查詢的設備 ID、時間范圍、零件 ID 等？是在 URL 參數中，還是在 POST 請求的 Body 中？Body 的格式是 JSON、XML 還是其他？
* **響應格式 (Response Format):** 真實系統返回的數據結構是怎樣的？與模擬的 JSON 結構是否一致？
2. 修改 Dify HTTP Request 節點:
* 定位工作流中所有調用模擬 API 的 HTTP Request 節點 (`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 不同。
* 因此，所有緊跟在 HTTP Request 節點之后的 `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 能力，創建更友好的用戶界面，允許用戶查詢特定設備狀態、歷史記錄或手動觸發分析。
8、項目資料包
有需要項目源碼的盆友可以付費后查看下方百度云盤鏈接（包含Code節點代碼）
