Dify 工作流全棧解析：從零構建你的 AI 應用流程引擎

AI工作流的導言：

? ? ? 工作流系統（Workflow System）是 Dify 的核心組成部分，它通過可視化編程界面支持創建復雜的 AI 應用程序。用戶可以將不同的功能塊連接起來，從而設計出用于處理數據、與 AI 模型交互、管理條件以及執行各種操作的工作流。

Dify 工作流的實現機制，通過分析代碼實現、數據流動和執行過程，充分理解工作流的實現原理。

1、工作流系統概述

1.1 核心概念
Dify 工作流系統是一個以圖（Graph）為基礎的執行引擎，它使用戶能夠通過可視化界面來設計和執行復雜的 AI 工作流。工作流由多種類型的節點（Node）構成，這些節點通過邊（Edge）相互連接，構成一個有向圖結構。

1.2 系統架構

工作流系統的核心構成模塊

1.2.1:?流程引擎（Workflow Engine）?
? ?作為整個工作流的“大腦”，負責解析流程定義，構建有向無環圖（DAG），并按規則調度和執行各個節點。它還控制節點之間的流轉邏輯與執行順序，確保流程按預期推進。

1.2.2:節點組件（Node Implementations）?
? ?提供多種功能類型的節點實現，如大語言模型調用（LLM）、數據庫查詢、知識庫檢索、條件判斷、代碼執行、HTTP 請求等，構成流程中的最小功能單元。

1.2.3:上下文與變量系統（Context & Variable Management）??
? ?負責在流程執行過程中管理數據的傳遞與共享，包括輸入參數、中間結果、輸出值以及全局/局部變量的作用域控制，確保節點間數據流動安全高效。

1.2.4:執行日志與監控（Execution Tracing & Monitoring）??
? ?記錄每次流程運行的詳細軌跡，包括每個節點的執行狀態、耗時、輸入輸出內容等信息，為調試、審計、性能優化和可視化提供數據支撐。

2、數據模型設計

2.1 工作流數據模型

Dify 采用多種模型來描述工作流及其運行狀況：

WorkflowModel：涵蓋工作流的基礎信息，如 ID、名稱、描述、配置等。
WorkflowRunModel：記錄工作流的運行情況，包括運行狀態、啟動時間、完成時間等。
WorkflowNodeExecutionModel：記錄節點的運行情況，包括節點類型、輸入、輸出、狀態等。
ConversationVariable：用于存儲會話變量，包含名稱、值類型、值等信息。
WorkflowDraftVariable：用于存儲草稿工作流中的變量，涵蓋會話變量、系統變量和節點變量。

2.2 工作流節點類型

Dify 工作流支持多種類型的節點，每種節點有不同的功能和配置：

Dify 提供了豐富多樣的節點類型，以滿足不同場景下的工作流構建需求：

START：作為工作流的入口點，標志著流程的開始。
END：表示工作流的終點，流程在此結束。
LLM：大語言模型節點，用于生成文本內容。
KNOWLEDGE_RETRIEVAL：知識檢索節點，用于從知識庫中檢索所需信息。
IF_ELSE：條件分支節點，根據設定條件選擇不同的執行路徑。
CODE：代碼執行節點，用于運行自定義代碼。
HTTP_REQUEST：HTTP 請求節點，用于與外部 API 進行交互。
TOOL：工具節點，用于調用預定義的工具。
AGENT：代理節點，用于執行復雜的任務。

3、工作流執行機制

3.1 工作流執行流程

初始化工作流運行記錄
?解析工作流配置，構建執行圖
從起始節點開始執行
根據圖的邊定義，確定下一個要執行的節點
執行節點，記錄執行結果
重復步驟 4-5，直到達到結束節點或出現錯誤
完成工作流執行，更新運行記錄

3.2 圖引擎執行機制

圖引擎作為工作流執行的核心組件，承擔著以下關鍵職責：

解析節點和邊配置：對工作流中定義的節點和邊的配置信息進行分析和處理。
構建邊映射和反向邊映射：創建邊的正向和反向映射關系，以便高效地管理和查詢節點之間的連接。
識別根節點和葉子節點：確定工作流的起始節點（根節點）和結束節點（葉子節點）。
檢查節點連接性和循環：驗證節點之間的連接是否正確，檢測是否存在循環依賴，確保工作流的執行邏輯合理。
管理并行執行：協調多個節點的并行執行，優化執行效率。
控制執行流程：根據工作流的定義和運行時條件，控制工作流的執行順序和流程走向。

4、變量管理機制

4.1 變量池設計
Dify 工作流通過變量池（VariablePool）來管理工作流執行過程中的變量。變量池涵蓋了以下幾類變量：

系統變量：以 `sys.` 為前綴，例如 `sys.query`（用戶輸入）、`sys.files`（用戶上傳的文件）。
環境變量：在工作流級別進行配置的變量。
會話變量：在會話中保持持久化的變量。
節點變量：各節點的輸入和輸出變量。

4.2 變量傳遞機制
在 Dify 工作流中，節點之間通過變量池進行數據傳遞。具體機制如下：

節點執行后：將輸出數據存儲到變量池中。
下一個節點執行時：從變量池中讀取所需的輸入變量。
支持變量引用：通過選擇器和模板字符串引用變量。
支持文件類型變量：允許傳遞文件類型的變量。
變量引用語法：使用 `{{#node_id.variable_name#}}` 的模板語法來引用變量。

5、節點實現機制

為了保證工作流中各類節點具備統一的執行接口，系統設計上采用面向對象的方式，讓所有節點繼承自一個共同的基類 —— BaseNode。每個節點必須實現自己的 _run 方法，作為其在流程中實際執行時的核心邏輯。

5.1 基礎節點結構

所有節點都繼承自?BaseNode?類，實現以下方法：

_run：節點的具體執行邏輯
_get_inputs：獲取節點的輸入變量
_get_outputs：處理節點的輸出變量

?這是?BaseNode?類的核心實現：

BaseNode 類詳解：所有節點的抽象基類

作為整個工作流系統中所有節點的統一基類，BaseNode 定義了節點運行所需的基礎結構和核心接口，確保各類節點在統一規范下執行。

1. 初始化方法（`init`）

構造函數接收必要的參數，包括節點唯一標識（node_id）、配置信息（config）、圖引擎引用（graph_engine）等，并據此初始化節點的基本屬性。這些屬性通常包括節點狀態、輸入輸出定義以及運行時所需的上下文環境。

2. 抽象執行方法（`_run`）

這是一個由子類必須實現的抽象方法，用于封裝節點的核心執行邏輯。每個具體類型的節點（如 LLM 節點、條件判斷節點等）都需要根據自身職責重寫該方法，返回執行結果或拋出異常。

3. 統一執行入口（`run`）

該方法是節點對外暴露的標準執行接口。它負責調用 _run 方法，并統一處理執行過程中的異常情況。最終，執行結果將被封裝為標準事件（Event）對象返回，供流程引擎進行后續處理或記錄。

通過這種設計，Dify 的工作流系統實現了高度可擴展的節點體系，使得新增節點類型變得簡單高效，同時保證了整體執行流程的一致性和可控性

5.2 節點類型實現

5.2.1 StartNode 實現

StartNode?是工作流的起始節點，負責將用戶輸入和系統變量作為節點的輸出：

StartNode?的實現非常簡單，它主要完成以下工作：

從變量池中獲取用戶輸入和系統變量
將系統變量添加到節點輸入中，以?SYSTEM_VARIABLE_NODE_ID.var?的形式作為鍵
返回包含這些輸入和輸出的?NodeRunResult，狀態為成功

5.2.2 IfElseNode 實現

IfElseNode?是條件分支節點，根據條件選擇執行路徑：

IfElseNode?的實現主要完成以下工作：

使用?ConditionProcessor?處理條件邏輯
遍歷?cases?結構中的條件組，并根據結果確定?selected_case_id
如果使用舊的結構，則調用?_should_not_use_old_function?進行兼容處理
返回包含條件結果的?NodeRunResult，并設置?edge_source_handle?以指示下一個要執行的節點

5.2.3 LLM 節點實現

作為工作流中的關鍵節點，LLM 節點主要承擔著調用大語言模型以生成文本的任務。

LLMNode?類的部分實現：

LLMNode?是一個典型的節點實現，負責調用大語言模型：

初始化節點參數和模型配置
處理輸入變量和文件
構建提示消息
調用 LLM 模型
處理模型返回的結果
生成節點執行結果

5.2.4 ToolNode 實現

ToolNode?是工具節點，負責調用預定義的工具：

ToolNode?的實現主要完成以下工作：

獲取工具信息和工具運行時
生成工具參數
獲取會話 ID
通過?ToolEngine.generic_invoke?調用工具
處理工具返回的結果
生成節點執行結果

5.2.5 KnowledgeRetrievalNode 實現

KnowledgeRetrievalNode?是知識檢索節點，負責從知識庫中檢索相關信息：

KnowledgeRetrievalNode?的實現主要完成以下工作：

從變量池中提取查詢變量
檢查查詢是否為空
進行速率限制檢查
定義檢索模型配置
執行知識檢索
處理檢索結果
生成節點執行結果

5.2.6 CodeNode 實現

CodeNode?是代碼執行節點，負責執行用戶定義的代碼：

CodeNode?的實現主要完成以下工作：

獲取代碼語言和代碼內容
從變量池中獲取輸入變量
通過?CodeExecutor.execute_workflow_code_template?執行代碼
檢查輸出變量的類型和長度
處理執行結果和潛在的異常
生成節點執行結果

5.2.7 AgentNode 實現

AgentNode?是代理節點，負責調用 AI 代理執行復雜任務：

AgentNode?的實現主要完成以下工作：

獲取代理策略
生成代理參數
獲取會話 ID
通過?strategy.invoke?調用代理
處理代理返回的結果
生成節點執行結果

5.2.8 HttpRequestNode 實現

HttpRequestNode?是 HTTP 請求節點，負責發送 HTTP 請求并處理響應：

HttpRequestNode?的實現主要完成以下工作：

獲取默認配置
初始化?HttpRequestExecutor
執行 HTTP 請求
處理響應（包括成功和失敗情況）
提取文件
生成節點執行結果

6、工作流數據流動?

6.1 工作流創建和發布

用戶在界面上設計工作流，定義節點和連接
系統將設計轉換為工作流配置
創建工作流模型和草稿變量
發布工作流，使其可被調用

6.2 工作流調試和執行

用戶觸發工作流執行
系統創建工作流運行記錄
圖引擎解析工作流配置，構建執行圖
按照圖的定義執行節點
記錄每個節點的執行狀態和結果
完成工作流執行，更新運行記錄

?7、圖引擎機制

圖引擎是工作流執行的核心，負責解析工作流圖結構并執行節點。

7.1 圖引擎實現

以下是?GraphEngine?類的部分實現：

GraphEngineThreadPool?是一個繼承自?ThreadPoolExecutor?的線程池，用于管理工作流的并行執行：

7.2 圖結構解析

圖引擎首先解析工作流的圖結構，包括：

解析節點：解析工作流中的所有節點，包括節點類型、配置等
解析邊：解析節點之間的連接關系，包括源節點、目標節點、源端口、目標端口等
構建節點映射：構建節點ID到節點對象的映射
構建邊映射：構建邊ID到邊對象的映射

7.3 圖結構實現

以下是?Graph?類的部分實現，它負責解析工作流配置并執行節點：

7.4 節點執行

圖引擎根據圖結構執行節點：

確定起始節點：通常是START節點
執行節點：調用節點的run方法
處理節點結果：根據節點執行結果確定下一個要執行的節點
處理并行執行：如果有多個分支，可以并行執行

7.4.1 節點執行的主要流程

節點執行的主要流程如下：

發出節點開始事件：觸發?NodeRunStartedEvent，通知系統節點開始執行
調用節點的?run?方法：執行節點的具體邏輯
處理節點事件：
- 處理?RunCompletedEvent：獲取節點執行結果
- 處理?RunStreamChunkEvent：處理流式輸出
- 處理?RunRetrieverResourceEvent：處理檢索資源
處理重試邏輯：如果節點執行失敗且配置了重試，則進行重試
更新變量池：將節點輸出變量添加到變量池中
發出節點完成事件：根據執行結果觸發相應事件
- 成功：觸發?NodeRunSucceededEvent
- 失敗：觸發?NodeRunFailedEvent
- 異常但繼續：觸發?NodeRunExceptionEvent
查找下一個要執行的節點：根據邊映射和條件確定下一個節點
執行下一個節點：可能是串行執行或并行執行

7.4.2 查找下一個節點的機制

在工作流執行過程中，確定下一個要執行的節點是關鍵步驟。GraphEngine?類的?_run?方法實現了這一機制：

1:獲取邊映射：通過?self.graph.edge_mapping.get(next_node_id)?獲取當前節點的所有出邊

2:單邊處理：如果只有一條出邊，直接獲取目標節點ID

3:多邊處理：如果有多條出邊，需要根據條件或并行策略確定下一個節點

? ? 條件分支：如果邊有運行條件，根據條件結果確定要執行的分支

并行分支：如果沒有條件或條件滿足，可能需要并行執行多個分支

4:并行分支執行：通過?_run_parallel_branches?方法處理并行分支

創建線程池和隊列管理并行執行
為每個分支創建一個線程執行
收集并處理所有分支的執行結果

5：檢查節點是否在當前并行分支內：確保節點執行不會跨越并行分支邊界?

通過這種機制，工作流系統能夠靈活地處理各種復雜的執行路徑，包括條件分支和并行執行，確保工作流按照設計的邏輯正確執行。

8、圖引擎與節點執行的通信機制

在工作流執行過程中，圖引擎與節點執行之間的通信是基于事件驅動機制來實現的。這種機制讓各個組件能夠以松耦合的方式進行交互，從而顯著提升了系統的可擴展性和可維護性。

8.1 事件驅動架構

工作流系統采用事件驅動架構，通過定義和傳遞各種事件來實現圖引擎與節點之間的通信。該架構具備以下優勢：

- 松耦合：圖引擎與節點之間通過事件進行交互，而非直接調用，從而降低了組件間的依賴性。
- 可擴展：新的節點類型和事件類型可以輕松集成到系統中，無需對現有代碼進行修改。
- 異步處理：事件可以異步處理，從而提高系統的響應性和吞吐量。
- 狀態追蹤：通過事件可以追蹤工作流的執行狀態和歷史記錄。

8.2 核心事件類型

工作流系統定義了多種事件類型，用于表示工作流執行過程中的不同狀態和操作：

8.2.1 圖級事件

GraphRunStartedEvent：工作流開始執行。
GraphRunSucceededEvent：工作流成功完成。
GraphRunFailedEvent：工作流執行失敗。
GraphRunPartialSucceededEvent：工作流部分成功（部分節點失敗但不影響整體結果）。

8.2.2 節點級事件

NodeRunStartedEvent：節點開始執行。
NodeRunSucceededEvent：節點執行成功。
NodeRunFailedEvent：節點執行失敗。
NodeRunExceptionEvent：節點執行異常但繼續執行。
NodeRunRetryEvent：節點重試執行。
NodeRunStreamChunkEvent：節點產生流式輸出。
NodeRunRetrieverResourceEvent：節點檢索資源。

8.2.3 并行分支事件

ParallelBranchRunStartedEvent：并行分支開始執行。
ParallelBranchRunSucceededEvent：并行分支執行成功。
ParallelBranchRunFailedEvent：并行分支執行失敗。

8.2.4 迭代和循環事件

IterationRunStartedEvent：迭代開始。
IterationRunNextEvent：迭代下一步。
IterationRunSucceededEvent：迭代成功完成。
IterationRunFailedEvent：迭代失敗。
LoopRunStartedEvent：循環開始。
LoopRunNextEvent：循環下一步。
LoopRunSucceededEvent：循環成功完成。
LoopRunFailedEvent：循環失敗。

8.3 事件傳遞機制

在工作流系統中，事件的傳遞遵循以下流程：

8.3.1事件生成：圖引擎或節點執行器生成事件：

8.3.2事件傳遞：通過 Python 生成器（Generator）機制傳遞事件：

8.3.3事件處理：工作流入口點（WorkflowEntry）接收事件并分發給回調處理器

8.3.4回調處理：回調處理器根據事件類型執行相應的操作

8.4 事件處理回調

工作流系統提供了回調接口，使得外部系統能夠注冊回調函數，以便處理工作流事件。

系統內置了多種回調實現，具體如下：

WorkflowLoggingCallback：用于記錄工作流的執行日志。
WorkflowAppRunnerCallback：用于處理應用層面的工作流事件。

8.5 事件與狀態管理

事件不僅用于通信，還用于管理工作流的狀態：

8.5.1節點狀態追蹤：通過事件記錄節點的執行狀態和結果

8.5.2變量傳遞：事件攜帶節點的輸入和輸出變量

8.5.3錯誤處理：事件攜帶錯誤信息，用于錯誤處理和重試

8.6 事件轉換與應用集成

工作流應用運行器（WorkflowAppRunner）負責將工作流事件轉化為應用層面的隊列事件，從而實現與應用系統的無縫集成。

這種轉換機制讓工作流系統在與外部應用系統實現無縫對接的同時，確保了自身內部實現的獨立性。

8.7 事件通信示例

以下是一個完整的事件通信流程示例，詳細展示了從節點執行到事件處理的全過程：

8.7.1 節點執行與事件產生

在圖引擎執行節點的過程中，會觸發并產生一系列事件：

8.7.2 事件傳遞與處理

事件通過工作流入口點傳遞給回調處理器：

8.7.3 回調處理器處理事件

回調處理器根據事件類型執行相應的操作：

8.7.4 應用運行器處理事件

應用運行器將工作流事件轉換為應用級別的隊列事件：

8.8 事件通信的優勢

基于事件的通信機制在圖引擎與節點執行器之間具有以下顯著優勢：

組件解耦：圖引擎與節點執行器通過事件進行交互，而非直接調用，從而降低了組件間的耦合度。
簡化調試：事件中包含完整的上下文信息，便于進行調試和問題排查。
支持異步執行：事件可以異步處理，支持并行執行和分布式部署。
可擴展性：新的節點類型和事件類型可以輕松集成到系統中，無需修改現有代碼。
狀態追蹤：通過事件可以完整記錄工作流的執行狀態和歷史，便于監控和審計。
錯誤處理：事件攜帶錯誤信息，支持靈活的錯誤處理策略和重試機制。

9、錯誤處理機制?

9.1 錯誤處理策略

工作流系統提供了兩種主要的錯誤處理策略：

FAIL_BRANCH：當節點執行失敗時，沿著失敗分支繼續執行。
- 將錯誤信息和類型添加到變量池中。
- 設置 edge_source_handle 為 FAILED，使工作流可以沿著專門處理失敗情況的分支繼續執行。
- 適用于需要針對失敗情況執行特定邏輯的場景。
DEFAULT_VALUE：當節點執行失敗時，使用預定義的默認值繼續執行。
- 將錯誤信息和類型添加到變量池中。
- 使用節點配置中預定義的默認值作為節點輸出。
- 適用于即使失敗也需要提供某種結果的場景。

9.2 節點重試機制

系統為某些類型的節點提供了執行失敗時的重試功能，具體如下：

重試配置：

max_retries：最大重試次數。
retry_interval_seconds：重試間隔時間（秒）。

重試流程：

節點執行失敗后，系統會檢查是否配置了重試。
如果當前重試次數小于最大重試次數，系統會觸發 NodeRunRetryEvent 事件。
系統等待指定的重試間隔時間。
系統重新執行節點。

重試事件：

系統會觸發 NodeRunRetryEvent 事件，該事件包含重試索引、開始時間等信息，可用于監控和記錄重試情況。

9.3 特殊錯誤處理行為的節點類型

系統定義了具有特定錯誤處理行為的節點類型，具體如下：

可繼續執行的節點類型（CONTINUE_ON_ERROR_NODE_TYPE）：
- 即使執行失敗，工作流仍可繼續執行。
- 例如：HTTP 請求節點、LLM 節點等。
- 這些節點可以配置錯誤策略（FAIL_BRANCH 或 DEFAULT_VALUE）。
可重試的節點類型（RETRY_ON_ERROR_NODE_TYPE）：
- 執行失敗時可以自動重試。
- 例如：HTTP 請求節點、數據庫操作節點等。
- 這些節點可以配置最大重試次數和重試間隔。

通過這些機制，工作流系統能夠靈活應對各種錯誤情況，從而提高工作流的健壯性和可靠性

10、變量管理機制

10.1 變量池

變量池是工作流中所有變量的集合，涵蓋了以下幾類：

用戶輸入變量：由用戶提供的輸入數據。
系統變量：由系統提供的變量，例如時間戳、會話ID等。
環境變量：與運行環境相關的變量。
會話變量：與當前會話相關的變量。
節點輸出變量：節點執行完成后產生的輸出變量。
以下是?VariablePool?類的部分實現：
class?VariablePool:
? ??"""
? ? Variable Pool
? ? """

? ??def?__init__(
? ? ? ? self,
? ? ? ? user_inputs: dict[str, Any],
? ? ? ? system_variables: dict[str, Any],
? ? ? ? environment_variables: dict[str, Any],
? ? ? ? session_variables: dict[str, Any],
? ? )?->?None:
? ? ? ??"""
? ? ? ? Initialize variable pool
? ? ? ? """
? ? ? ? self.user_inputs = user_inputs
? ? ? ? self.system_variables = system_variables
? ? ? ? self.environment_variables = environment_variables
? ? ? ? self.session_variables = session_variables

? ? ? ??# Initialize variable dictionary
? ? ? ? self.variable_dictionary: dict[str, Any] = {}

? ??def?add(self, node_id: str, variable_name: str, variable_value: Any)?->?None:
? ? ? ??"""
? ? ? ? Add variable to variable pool
? ? ? ? """
? ? ? ??# Check if variable value is File
? ? ? ??if?isinstance(variable_value, File):
? ? ? ? ? ??# Convert File to dict
? ? ? ? ? ? variable_value = variable_value.to_dict()

? ? ? ??# Add variable to variable dictionary
? ? ? ? self.variable_dictionary[f"{node_id}.{variable_name}"] = variable_value

? ??def?get_variable(self, variable_selector: str)?-> Any:
? ? ? ??"""
? ? ? ? Get variable from variable pool
? ? ? ? """
? ? ? ??# Check if variable selector is empty
? ? ? ??ifnot?variable_selector:
? ? ? ? ? ??returnNone

? ? ? ??# Check if variable selector is system variable
? ? ? ??if?variable_selector.startswith(SYSTEM_VARIABLE_NODE_ID):
? ? ? ? ? ??# Get system variable
? ? ? ? ? ? variable_name = variable_selector.split(".",?1)[1]
? ? ? ? ? ??return?self.get_system_variable(variable_name)

? ? ? ??# Check if variable selector is user input
? ? ? ??if?variable_selector.startswith(USER_INPUT_NODE_ID):
? ? ? ? ? ??# Get user input
? ? ? ? ? ? variable_name = variable_selector.split(".",?1)[1]
? ? ? ? ? ??return?self.get_user_input(variable_name)

? ? ? ??# Check if variable selector is environment variable
? ? ? ??if?variable_selector.startswith(ENVIRONMENT_VARIABLE_NODE_ID):
? ? ? ? ? ??# Get environment variable
? ? ? ? ? ? variable_name = variable_selector.split(".",?1)[1]
? ? ? ? ? ??return?self.get_environment_variable(variable_name)

? ? ? ??# Check if variable selector is session variable
? ? ? ??if?variable_selector.startswith(SESSION_VARIABLE_NODE_ID):
? ? ? ? ? ??# Get session variable
? ? ? ? ? ? variable_name = variable_selector.split(".",?1)[1]
? ? ? ? ? ??return?self.get_session_variable(variable_name)

? ? ? ??# Get variable from variable dictionary
? ? ? ??return?self.variable_dictionary.get(variable_selector)