Chapter 4: 工作流 (Workflow)

在上一章 流程 (Flow) 中，我們學習了如何將多個節點 (Node) 串聯起來，形成一個有順序、有邏輯的執行路徑。我們看到了「流程」就像一位生產線經理，指揮著各個節點按部就班地工作。

現在，我們將更進一步，探討如何運用這些知識來解決更大型、更複雜的真實世界任務。這就是「工作流 (Workflow)」概念的用武之地。想像一下，您不再只是管理一條生產線，而是要設計並運營整個工廠的生產藍圖。

為何需要工作流？——將宏大藍圖化為具體步驟

許多真實世界的任務都非常複雜，遠非單一的指令或一個簡單的流程 (Flow) 就能完成。例如，要撰寫一本完整的書籍，涉及的步驟可能包括：

1. 構思與大綱：確定主題、目標讀者、核心內容，並擬定詳細的書籍大綱。
2. 初稿撰寫：根據大綱，逐章撰寫內容。
3. 內容審閱：邀請編輯或同儕審閱初稿，提供修改意見。
4. 修改與潤飾：根據回饋進行修改，調整結構，潤飾文字。
5. 排版與校對：進行專業排版，並仔細校對，消除錯誤。
6. 出版：最終定稿並出版。

如果試圖一次性完成所有這些，不僅容易出錯，而且難以管理進度。更明智的方法是將這個龐大的專案「分解」成一系列較小、可管理且相互關聯的階段。

工作流 (Workflow) 正是這樣一種理念與實踐：它指將一個複雜任務分解為一系列較小、可管理的步驟（在 PocketFlow 中通常是節點 (Node)），並將它們串聯起來形成一個有序的執行流程。PocketFlow 的核心設計便是支持這種任務分解，通過節點間的鏈接和行動指令來定義工作流的具體執行路徑，確保任務按部就班地完成。

圖示：工作流將複雜任務分解為節點鏈

在本章中，我們將學習如何運用 PocketFlow 的組件來設計和實現這樣的工作流。

工作流在 PocketFlow 中的體現

您可能會問，PocketFlow 中是否有一個叫做 Workflow 的特定類別呢？

實際上，在 PocketFlow 目前的設計中，「工作流 (Workflow)」更多的是一種設計模式或一種方法論，而不是一個獨立的類別。我們通過組合使用已經學過的 節點 (Node) 和 流程 (Flow) 來構建和實現工作流。

• 任務分解：將複雜任務拆解成多個獨立的、功能單一的節點 (Node)。
• 流程編排：使用 流程 (Flow) 將這些節點按照邏輯順序和條件分支串聯起來。
• 數據流動：利用 共享儲存 (Shared Store) 在節點之間傳遞數據。

因此，當我們談論在 PocketFlow 中創建一個「工作流」時，我們實質上是在設計一個結構良好、目標明確的 流程 (Flow)，用以解決一個特定的複雜問題。

小試身手：打造一個「文章撰寫工作流」

讓我們以一個常見的任務為例：自動撰寫一篇關於特定主題的文章。這個任務可以分解為以下幾個主要步驟：

1. 產生大綱 (Generate Outline)：根據給定主題，產生文章的結構大綱。
2. 撰寫內容 (Write Content)：根據大綱，為每個部分撰寫詳細內容。
3. 風格潤飾 (Apply Style)：將撰寫的內容統一調整為特定的風格（例如：更口語化、更專業等）。

我們將為每個步驟創建一個節點 (Node)，然後將它們組合成一個流程 (Flow)，這個流程就構成了我們的「文章撰寫工作流」。

步驟 1：定義節點

首先，我們定義三個節點。為了簡化，我們將用模擬函數 call_llm_for_task 來代表對大型語言模型的調用。

from pocketflow import Node, Flow

# 模擬呼叫大型語言模型
def call_llm_for_task(task_description, input_data):
    print(f"🤖 正在為任務 '{task_description}' 呼叫 LLM，輸入：'{str(input_data)[:50]}...'")
    if task_description == "產生大綱":
        return f"主題 '{input_data}' 的大綱：1. 引言 2. 主要論點 3. 結論"
    elif task_description == "撰寫內容":
        return f"基於大綱 '{input_data[:30]}...' 的詳細內容。"
    elif task_description == "風格潤飾":
        return f"已潤飾的文章：'{input_data[:30]}...' (風格：引人入勝)"
    return "未知任務的 LLM 結果"

class GenerateOutlineNode(Node):
    def prep(self, shared):
        return shared.get("topic", "未知主題") # 從共享儲存讀取主題

    def exec(self, prep_res): # prep_res 是主題
        return call_llm_for_task("產生大綱", prep_res)

    def post(self, shared, prep_res, exec_res): # exec_res 是大綱
        shared["outline"] = exec_res
        print(f"✅ 大綱已產生並儲存：{exec_res}")

class WriteContentNode(Node):
    def prep(self, shared):
        return shared.get("outline", "沒有大綱") # 從共享儲存讀取大綱

    def exec(self, prep_res): # prep_res 是大綱
        return call_llm_for_task("撰寫內容", prep_res)

    def post(self, shared, prep_res, exec_res): # exec_res 是初稿
        shared["draft_content"] = exec_res
        print(f"✅ 初稿已撰寫並儲存：{str(exec_res)[:60]}...")

class ApplyStyleNode(Node):
    def prep(self, shared):
        return shared.get("draft_content", "沒有初稿") # 從共享儲存讀取初稿

    def exec(self, prep_res): # prep_res 是初稿
        return call_llm_for_task("風格潤飾", prep_res)

    def post(self, shared, prep_res, exec_res): # exec_res 是最終文章
        shared["final_article"] = exec_res
        print(f"🎉 最終文章已完成並儲存：{str(exec_res)[:60]}...")

• GenerateOutlineNode：接收一個「主題」(topic)，輸出一個「大綱」(outline)。
• WriteContentNode：接收「大綱」，輸出「初稿內容」(draft_content)。
• ApplyStyleNode：接收「初稿內容」，輸出「最終文章」(final_article)。

每個節點都遵循 prep -> exec -> post 的模式，並使用共享儲存 (Shared Store)來讀取輸入和儲存輸出。

步驟 2：將節點連接成流程 (即工作流)

現在，我們將這些節點實例化，並將它們連接成一個流程 (Flow)。由於這是一個簡單的線性序列，我們使用 >> 運算符進行連接。

# 建立節點實例
outline_node = GenerateOutlineNode()
write_node = WriteContentNode()
style_node = ApplyStyleNode()

# 將節點連接起來形成流程
# 假設每個節點成功後都執行下一個 (使用預設的 "default" 行動)
outline_node >> write_node
write_node >> style_node

# 建立流程，將 outline_node 設為起始節點
article_workflow = Flow(start=outline_node)

這段程式碼定義了節點的執行順序：outline_node 完成後執行 write_node，write_node 完成後執行 style_node。這個 article_workflow 就是我們設計的「文章撰寫工作流」。

步驟 3：運行工作流

最後，我們可以準備初始的共享儲存 (Shared Store)（包含文章主題），然後運行整個工作流。

# 準備共享儲存，放入初始資料
shared_data = {"topic": "AI 的未來"}

print(f"=== 🚀 開始文章撰寫工作流，主題：{shared_data['topic']} ===")

# 運行工作流
article_workflow.run(shared_data)

print("\n=== ✨ 工作流執行完畢 ===")
print(f"最終文章預覽：{shared_data.get('final_article')}")

當您運行這段程式碼時，您會看到類似以下的輸出（順序和具體內容可能因模擬函數而略有不同）：

=== 🚀 開始文章撰寫工作流，主題：AI 的未來 ===
🤖 正在為任務 '產生大綱' 呼叫 LLM，輸入：'AI 的未來...'
✅ 大綱已產生並儲存：主題 'AI 的未來' 的大綱：1. 引言 2. 主要論點 3. 結論
🤖 正在為任務 '撰寫內容' 呼叫 LLM，輸入：'主題 'AI 的未來' 的大綱：1. 引言 2. 主...'
✅ 初稿已撰寫並儲存：基於大綱 '主題 'AI 的未來' 的大綱：1. 引言 2. 主...' 的詳細內容。
🤖 正在為任務 '風格潤飾' 呼叫 LLM，輸入：'基於大綱 '主題 'AI 的未來' 的大綱：1. 引言 2. 主...'
🎉 最終文章已完成並儲存：已潤飾的文章：'基於大綱 '主題 'AI 的未來' 的大綱：1. 引言 2. 主...' (風格：引人入勝)

=== ✨ 工作流執行完畢 ===
最終文章預覽：已潤飾的文章：'基於大綱 '主題 'AI 的未來' 的大綱：1. 引言 2. 主...' (風格：引人入勝)

正如您所見，我們的「文章撰寫工作流」成功地按照預期順序執行了所有步驟，並將最終結果儲存在了共享儲存 (Shared Store)中。我們通過將複雜任務分解為簡單節點，並用流程將它們串聯起來，有效地實現了一個自動化工作流。

設計有效工作流的考量

在設計自己的工作流時，有幾個重要的原則可以幫助您：

1. 任務分解的粒度 (Granularity of Task Decomposition)：

   • 不宜過粗 (Not too coarse)：如果一個節點承擔了太多複雜的邏輯，它本身可能難以實現、測試和維護。而且，如果其中一小部分失敗，整個大節點都需要重試。
   • 不宜過細 (Not too granular)：如果節點拆分得過細，例如每個節點只做一件微不足道的小事，可能會導致流程過於冗長，節點間的數據傳遞變得繁瑣，並且 LLM 可能因為上下文不足而導致各節點結果不一致。
   • 尋找平衡點 (Find the sweet spot)：目標是找到一個平衡點，使得每個節點都執行一個有意義、內聚的子任務。這通常需要通過幾次迭代和實驗來找到。

2. 明確的職責 (Clear Responsibilities)：

   • 每個節點 (Node) 都應該有一個清晰、單一的職責。這使得工作流更容易理解和修改。

3. 數據流的設計 (Data Flow Design)：

   • 仔細規劃數據如何在節點之間通過共享儲存 (Shared Store)流動。明確每個節點需要什麼輸入，以及它會產生什麼輸出。

4. 錯誤處理與容錯 (Error Handling and Fault Tolerance)：

   • 考慮工作流中哪些步驟可能會失敗，並利用節點的重試機制 (max_retries, wait) 和回退邏輯 (exec_fallback) 來增強工作流的強韌性。

5. 可重用性 (Reusability)：

   • 設計可重用的節點和子流程。如果某個子任務（例如：從文件中讀取文本）在多個工作流中都會用到，可以將其封裝成一個獨立的節點或子流程 (Flow)，以便在不同地方重複使用。

最佳實踐提醒：

• 任務劃分不宜過粗，否則單個 LLM 調用可能難以處理。
• 任務劃分不宜過細，否則 LLM 可能缺乏足夠上下文，導致節點間結果不一致。
• 通常需要多次迭代才能找到最佳平衡點。
• 如果任務包含大量動態分支或需要根據上下文靈活決策的邊緣情況，可以考慮使用更高級的抽象，例如我們將在後續章節介紹的 代理人/智能體 (Agent)。 {: .best-practice }

工作流的內部運作回顧

由於我們在 PocketFlow 中使用 流程 (Flow) 來實現工作流，因此工作流的內部運作機制與流程的運作機制是相同的。這裡我們快速回顧一下：

1. 啟動：當您調用 workflow.run(shared_data)（其中 workflow 是一個 Flow 物件實例）時，流程從其指定的起始節點開始。
2. 節點執行：
   • 當前節點執行其 prep、exec（可能包含重試或回退）和 post 方法。
   • post 方法回傳一個「行動指令」。
3. 流程導航：
   • 流程根據當前節點回傳的「行動指令」以及預先定義的轉換規則，決定下一個要執行的節點。
4. 迭代：重複步驟 2 和 3，直到沒有下一個節點可執行，或者某個節點的行動指令沒有對應的轉換規則為止。
5. 數據共享：所有節點都通過同一個共享儲存 (Shared Store)實例來讀取和寫入數據，從而實現數據在工作流中的流動。

下面的序列圖展示了我們的「文章撰寫工作流」執行時的簡化交互過程：

這個圖清晰地展示了控制流（節點如何被依次調用）和數據流（數據如何在節點之間通過共享儲存傳遞）。更詳細的關於流程 (Flow) 的內部機制，您可以參考上一章的內容。

總結

在本章中，我們探討了「工作流 (Workflow)」的概念及其在 PocketFlow 中的實現：

• 工作流是將複雜任務分解為一系列有序、可管理的步驟（節點）並將其串聯起來的執行流程。
• 在 PocketFlow 中，工作流通常是通過精心設計的 流程 (Flow) 來實現的，它利用 節點 (Node) 進行任務執行，並通過 共享儲存 (Shared Store) 進行數據傳遞。
• 我們通過一個「文章撰寫工作流」的例子，展示了如何定義節點、將它們組合成流程，並運行整個工作流。
• 有效的任務分解粒度、清晰的節點職責和良好的數據流設計是構建成功的的工作流的關鍵。

掌握了工作流的設計理念，您就能夠運用 PocketFlow 來解決更廣泛、更複雜的自動化問題，將宏大的想法一步步變為現實。

在我們設計好單個工作流之後，經常會遇到需要對大量不同的輸入數據重複執行同一個工作流的情況。例如，我們可能需要為 100 個不同的主題都生成一篇文章。如何高效地處理這種情況呢？這就是我們下一章將要介紹的內容：批次處理 (Batch)。