流程引擎 SpiffWorkflow

SpiffWorkflow 是流程引擎，這裡的「流程」有點含糊，在談 SpiffWorkflow 前，先來說說它的流程是哪種流程。

流程

流程的英文可以是 process 或 workflow，或者有時候也常常用更簡化的形式「flow」表示，這些詞彙混用的狀況時常有之，我們也都概稱為流程，對於字面上的意思或許可以不深究，但對於應用面可就要計較一番了，對於流程，比較常見到的有：

• UI flow，這是指用戶的操作流程，通常在 UX 領域可見，典型的應用是 Penpot、Figma 這類設計應用。
• Data flow，這是指資料流程，目前比較常用於表示大數據的 ETL 流程。
• Business process，這是指商業流程，也是用戶最能具體感受到的流程，包括人事、總務、採購、業務等各類申辦流程，這類產品又稱為 BPM（business process management）。

其中後兩者比較容易搞混，如果你想搭建的是申辦、簽核系統，那就應該找 business process 方案，如果你想搭建的是資料流處理系統，那就應該找 data flow 方案。

對於 data flow，比較著名的工具有 Apache Airflow，它是以 DAG 為概念的資料流程執行工具，它的設計模式決定了它的使用姿勢，它當然可以有分支也當然可以有人為決策，只是相較於專門的 business process 方案，要跑平行簽核、多重指派，在 Airflow 要實現這些得花更多力氣罷了，就像你不會在 PowerPoint 寫文件，也不會在 Word 寫程式一樣，不是不能，只是不適合。

而 SpiffWorkflow 就是專門針對 business process 的流程引擎，同類型的引擎或產品有這些：

• 一等一科技 BPM，他們家的旗艦產品是 UOF EIP，BPM 是其中主要模組。這裡推一下一等一科技，他們家的導入、客服、技術服務都很有制度也很專業。
• 新人類資訊 FlowMaster BPM，牌價比樓上貴一點。
• 華苓科技 Agentflow BPM。
• 慧智科技 SmartBPM.NET。

上面這些是國內的，還有很多鼎新什麼的，就不幫它們打廣告了。這些國內的產品都是完整的 BPM 解決方案，有流程引擎、有表單設計、有人員組織、有統計圖表等等，也有面向開發端的 API。

國外的則有：

• Camnuda，應該是國外最大的 BPM 吧。
• ProcessMaker，除了 Camuda，商品化最成熟的產品。
• jBPM，紅帽主導的 BPM。
• Flowable，德國的。
• Alfresco Process Services，做 EIP 的 Alfresco 也有 BPM 產品。
• 馳騁 BPM，中國的 BPM。
• Bonita。
• Imixs Workflow。
• Automatiko。
• SpiffWorkflow。

國外產品比較偏向終端用戶的大概有 Cammda、ProcessMaker、馳騁 BPM、Flowable，其它則是比較面向開發者，包括本文的主角 SpiffWorkflow。

BPMN

上面這些 BPM 的共同點是它們都以 BPMN 為基礎，BPMN 全稱 Business Process Model and Notation，是專門用來表示商業流程的一種圖，相較於一般的流程圖或 DAG，BPMN 有更豐富的符號來表示一段商業流程，下面是一張核彈發射流程的 BPMN：

在 BPMN 的符號中：

• 圓圈表示 event，這裡有 Start event 和 End event。
• 菱形表示 gateway，簡單理解就是叉路，將軍可以決定要不要射，總統也可以決定要不要射。
• 圓角方框表示 task，就是具體要做的事情。

雖然任何一款繪圖應用都可以畫出這樣的圖，但要讓程式使用，就必須用標準組織制定的機讀格式製作 BPMN，目前 BPMN 的機讀格式標準由 OMG 機構維護，本質上它是一種 XML 文件，前面那些國外的 BPM 大多可以讀入標準的 BPMN 檔案，而國內的 BPM 大多是玩自己的一套，這也沒什麼不好，只要能跑流程就好。

對於 BPMN 的認識，簡單帶過就好，因為下面 SpiffWorkflow 我們會跳過 BPMN 的部份 :p。

SpiffWorkflow

SpiffWorkflow 是 business process 的執行引擎，具體一點說，它本質上是狀態機與規則引擎的綜合體，流程制定好後，SpiffWorkflow 負責告訴我們當前該流程內所有工項的狀態，有的可能是 COMPLETED，有的可能是 READY，有的可能是 MAYBE、FUTURE 等等，隨著流程的推進，每個工項的狀態也會隨之變更，這也是 SpiffWorkflow 的主要價值所在，試想如果要自行判斷複雜流程的狀態，我們要寫多少規則才可能實現？

以上面的射核彈流程為例：

SpiffWorkflow 可以讀入像這樣的 BPMN 檔案，並作為 workflow spec 之用，在 SpiffWorkflow 的設計裡：

• 制定的流程稱為 workflow spec
• 真正執行的流程稱為 workflow instance

一份 spec 之下會有許多的 instance，核彈一號的與核彈二號有各自的 workflow instance，也有各自的流程走向與結局，但它們的 spec 是相同的，都是將軍先決定，總統再決定。

制定 Workflow Spec

前面提到，SpiffWorkflow 可以讀入 BPMN 檔案並解析成 workflow spec，但這裡我們不這麼做，我們玩硬派的，手刻 spec，程式碼如下：

from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from SpiffWorkflow.specs.ExclusiveChoice import ExclusiveChoice
from SpiffWorkflow.specs.Simple import Simple
from SpiffWorkflow.specs.Cancel import Cancel

from SpiffWorkflow.operators import Equal, Attrib

def my_nuclear_strike(msg):
    print("Launched:", msg)

class NuclearStrikeWorkflowSpec(WorkflowSpec):
    def __init__(self):
        super().__init__()

        # The first step of our workflow is to let the general confirm
        # the nuclear strike.
        general_choice = ExclusiveChoice(wf_spec=self, name='general')
        self.start.connect(taskspec=general_choice)

        # The default choice of the general is to abort.
        cancel = Cancel(wf_spec=self, name='workflow_aborted')
        general_choice.connect(task_spec=cancel)

        # Otherwise, we will ask the president to confirm.
        president_choice = ExclusiveChoice(wf_spec=self, name='president')
        cond = Equal(Attrib(name='confirmation'), 'yes')
        general_choice.connect_if(condition=cond, task_spec=president_choice)

        # The default choice of the president is to abort.
        president_choice.connect(task_spec=cancel)

        # Otherwise, we will perform the nuclear strike.
        strike = Simple(wf_spec=self, name='nuclear_strike')
        president_choice.connect_if(condition=cond, task_spec=strike)

        # Now we connect our Python function to the Task named 'nuclear_strike'
        strike.completed_event.connect(callback=my_nuclear_strike)

        # As soon as all tasks are either "completed" or  "aborted", the
        # workflow implicitely ends.

可以看到，SpiffWorkflow.specs 模組下有許多用於定義 spec 的 class，這邊我們用到其中三個：

• Simple：表示最單純的 task。
• ExclusiveChoice：表示單選的 gateway。
• Cancel：表示 end，即取消發射。

其中 ExclusiveChoice 預設是走向 Cancel，除非將軍與總統都把 confirmation 設為 yes，這部份用到 SpiffWorkflow.operators 的運算子 class Equal 與 Attrib，整段流程不複雜，多看兩下應該可以參透，至於那 confirmation 要怎麼設、流程具體怎麼跑，讓我們看下去。

建立與操控 Workflow Instance

建立 workflow instance：

from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.task import Task, TaskState

workflow_instance = Workflow(workflow_spec=NuclearStrikeWorkflowSpec())

Instance 建立之後，就可以開始把玩了，在跑流程之前，先玩轉 instance 的一些方法。

Dump

workflow_instance.dump()

輸出如下：

358155a6-4776-46f2-aed2-cae4b18d6391/0: Task of Root State: COMPLETED Children: 1
  a18df1d8-c5a2-4bd0-832b-39b9d38a46f3/0: Task of Start State: READY Children: 1
    4d0eb169-15eb-42f5-a895-359590b5ce69/0: Task of general State: FUTURE Children: 2
      32726718-f032-417c-8325-bdc2db2f878d/0: Task of workflow_aborted State: LIKELY Children: 0
      41f2cb59-5d15-4735-b0af-fb42d75a1e0c/0: Task of president State: MAYBE Children: 2
        821401fd-a55d-45a4-8131-fa6af712b6d9/0: Task of workflow_aborted State: MAYBE Children: 0
        150966e3-05d2-4463-9209-d6506fa8f834/0: Task of nuclear_strike State: MAYBE Children: 0

dump() 即把流程的狀態以父子階層的格式秀出來，這裡 SpiffWorkflow 把 event、gateway、task 都以 task 概稱之，一筆 task 有以下欄位：

• 最前面是它的 UUID，每個 spec 的 instance 的 task 的 UUID 都是獨立的。
• 中間是 task 名。
• 後面接 task 狀態，COMPLETED、READY 等等。
• 最後為旗下子 task 的數量。

Get tasks

workflow_instance.get_tasks()

輸出如下：

[<Task object (Root) in state COMPLETED at 0x7f9405f4f9d0>,
 <Task object (Start) in state READY at 0x7f9405f4d270>,
 <Task object (general) in state FUTURE at 0x7f9405f4cdc0>,
 <Task object (workflow_aborted) in state LIKELY at 0x7f9405f4e350>,
 <Task object (president) in state MAYBE at 0x7f9405f4d2d0>,
 <Task object (workflow_aborted) in state MAYBE at 0x7f9405f4fa30>,
 <Task object (nuclear_strike) in state MAYBE at 0x7f9405f4e3b0>]

get_tasks() 也是展示當前狀態，以 Python 原生 Task list 的形式表現。

Get tasks from spec name

workflow_instance.get_tasks_from_spec_name(name='general')

輸出如下：

[<Task object (general) in state FUTURE at 0x7f9405f4cdc0>]

以 task 名稱取得 task 物件，注意，這裡拿到的是 task list。

拿到 Task 物件後，也有一些方法與屬性。

Task 方法與屬性

以將軍的發射決策 task 為例：

general_task: Task = workflow_instance.get_tasks_from_spec_name(name='general')[0]

ID

general_task.id

輸出如下：

UUID('4d0eb169-15eb-42f5-a895-359590b5ce69')

狀態

general_task.get_state_name()

輸出如下：

'FUTURE'

對於 task 狀態，SpiffWorkflow 有以下幾種：

• MAYBE、LIKELY，這類屬於預期（PREDICTED）會發生的。
• FUTURE、WAITING、READY、STARTED，這類屬於確定（DEFINITE）會發生。
• COMPLETED、ERROR、CANCELLED，這類屬於結束（FINISHED）的。

隨著流程的推進，前面的決策會影響後面工項的狀態，這裡體現了 SpiffWorkflow 作為狀態機的特性。

雖然狀態這麼多，但最重要的是 READY，它表示接著要執行的 task，也是我們在使用 SpiffWorkflow 流程引擎時主要關注的狀態。

至於每種狀態的意義，除了望文生義外，也可以參見〈SpiffWorkflow Concepts〉。

資料

前面提到，將軍與總統都要做出決斷，把 confirmation 設為 yes 才可發射核彈，而這 confirmation 就會以 task data 的形式存在，做好決斷，推進流程，範例如下：

# 跑前面的流程

general_task.set_data(confirmation='yes') # 做決斷
workflow_instance.run_task_from_id(task_id=general_task.id) # 跑流程

這樣決斷一波下來，總統的決斷 task 就會變成 READY，反之，若是不設定 confirmation 為 yes，那就變成 cancel task 是 READY 囉，總統就變成 CANCELLED。

認識完 instance 與 task，來重頭跑一次流程吧！

跑流程

建立 workflow instance，檢視初始狀態：

workflow_instance = Workflow(workflow_spec=NuclearStrikeWorkflowSpec())
workflow_instance.dump()

d6ccb20e-69d9-4a14-9d01-4563a0addc2f/0: Task of Root State: COMPLETED Children: 1
  ebd7018d-d190-4a3b-a4c1-02877006b77e/0: Task of Start State: READY Children: 1
    b6917bf4-9efa-4f86-8025-2edbc5f766f8/0: Task of general State: FUTURE Children: 2
      02a4cb8d-994b-4892-8ef5-53af2791b2e2/0: Task of workflow_aborted State: LIKELY Children: 0
      c38a78d5-c24b-4123-a00c-772f76303134/0: Task of president State: MAYBE Children: 2
        1949bc90-309e-404f-94fd-332693c1aa91/0: Task of workflow_aborted State: MAYBE Children: 0
        8c518cb2-637d-4db6-85be-867039d06481/0: Task of nuclear_strike State: MAYBE Children: 0

那個 Root 和 Start task 是 SpiffWorkflow 預設的，來跑 Start task：

start_tasks: list[Task] = workflow_instance.get_tasks_from_spec_name(name='Start')
for task in start_tasks:
   if task.state == TaskState.READY:
       workflow_instance.run_task_from_id(task_id=task.id)
workflow_instance.dump()

d6ccb20e-69d9-4a14-9d01-4563a0addc2f/0: Task of Root State: COMPLETED Children: 1
 ebd7018d-d190-4a3b-a4c1-02877006b77e/0: Task of Start State: COMPLETED Children: 1
   b6917bf4-9efa-4f86-8025-2edbc5f766f8/0: Task of general State: READY Children: 2
     02a4cb8d-994b-4892-8ef5-53af2791b2e2/0: Task of workflow_aborted State: LIKELY Children: 0
     c38a78d5-c24b-4123-a00c-772f76303134/0: Task of president State: MAYBE Children: 2
       1949bc90-309e-404f-94fd-332693c1aa91/0: Task of workflow_aborted State: MAYBE Children: 0
       8c518cb2-637d-4db6-85be-867039d06481/0: Task of nuclear_strike State: MAYBE Children: 0

可以看到，Start task 變成 COMPLETED、general task 變成 READY。

接著跑 general task，讓他射：

general_tasks: list[Task] = workflow_instance.get_tasks_from_spec_name(name='general')
for task in general_tasks:
    if task.state == TaskState.READY:
        task.set_data(confirmation='yes')
        workflow_instance.run_task_from_id(task_id=task.id)
workflow_instance.dump()

d6ccb20e-69d9-4a14-9d01-4563a0addc2f/0: Task of Root State: COMPLETED Children: 1
  ebd7018d-d190-4a3b-a4c1-02877006b77e/0: Task of Start State: COMPLETED Children: 1
    b6917bf4-9efa-4f86-8025-2edbc5f766f8/0: Task of general State: COMPLETED Children: 1
      c38a78d5-c24b-4123-a00c-772f76303134/0: Task of president State: READY Children: 2
        1949bc90-309e-404f-94fd-332693c1aa91/0: Task of workflow_aborted State: LIKELY Children: 0
        5541d692-1c42-4872-a8d6-22fba2491761/0: Task of nuclear_strike State: MAYBE Children: 0

現在 president task 變成 READY，繼續讓總統射：

president_tasks: list[Task] = workflow_instance.get_tasks_from_spec_name(name='president')
for task in president_tasks:
    if task.state == TaskState.READY:
        task.set_data(confirmation='yes')
        workflow_instance.run_task_from_id(task_id=task.id)
workflow_instance.dump()

d6ccb20e-69d9-4a14-9d01-4563a0addc2f/0: Task of Root State: COMPLETED Children: 1
  ebd7018d-d190-4a3b-a4c1-02877006b77e/0: Task of Start State: COMPLETED Children: 1
    b6917bf4-9efa-4f86-8025-2edbc5f766f8/0: Task of general State: COMPLETED Children: 1
      c38a78d5-c24b-4123-a00c-772f76303134/0: Task of president State: COMPLETED Children: 1
        5541d692-1c42-4872-a8d6-22fba2491761/0: Task of nuclear_strike State: READY Children: 0

現在都確認了，核彈進入戰術位置，射：

strike_tasks: list[Task] = workflow_instance.get_tasks_from_spec_name(name='nuclear_strike')
for task in strike_tasks:
    if task.state == TaskState.READY:
        workflow_instance.run_task_from_id(task_id=task.id)
workflow_instance.dump()

Launched: <SpiffWorkflow.workflow.Workflow object at 0x7fb0b09e30d0>
d6ccb20e-69d9-4a14-9d01-4563a0addc2f/0: Task of Root State: COMPLETED Children: 1
  ebd7018d-d190-4a3b-a4c1-02877006b77e/0: Task of Start State: COMPLETED Children: 1
    b6917bf4-9efa-4f86-8025-2edbc5f766f8/0: Task of general State: COMPLETED Children: 1
      c38a78d5-c24b-4123-a00c-772f76303134/0: Task of president State: COMPLETED Children: 1
        5541d692-1c42-4872-a8d6-22fba2491761/0: Task of nuclear_strike State: COMPLETED Children: 0

在這個例子中，將軍、總統的決斷都是直接設定就跑了，當然在更實際的應用中，SpiffWorkflow 作為流程引擎，還需要外部的程式與前端配合，才能真正讓用戶去輸入他的決斷，這部份就取決於各自的應用看要怎麼發揮了。

另外 SpiffWorkflow 欠缺一些完整的商業 BPM 產品的特性，例如權限（哪個節點歸誰簽）、代簽、會簽、組織、組織主管等，所以它真的只是流程引擎，適合應用的場景是自有專案又需要內含稍微複雜的商業流程，而不適合拿它來搭建一套請假系統、出差申請系統、請購系統等等，這些往往都牽涉到完整的組織與主管，遇到這類需求還是找商業化的 BPM 產品吧。

Share