CrewAI Flow 入門：用狀態機把多個 Crew 串成可控流程

當你的流程從「研究 + 寫報告」演變成「研究 → 產文 → 品質審核 → 通過就發布、不通過就重跑」，單一 Crew 就不夠了。

不是 Crew 不好，而是它本來就不是設計來處理「有條件分支」的情境。單一 Crew 的任務流程是線性的，而真實業務邏輯通常長這樣：「做完 A，判斷結果，決定要跑 B 還是跑 C」。

Flow 就是 CrewAI 用來處理這種情境的工具——它像一個總控台，負責協調多個 Crew 的執行順序和條件邏輯。

Flow 在解決什麼問題？

三個核心能力：

管理跨步驟狀態：資料可以從第一個步驟一路傳到最後一個步驟，不用靠全域變數
條件分支：根據某個條件決定走哪條路，而不是全部跑完
多 Crew 協調：把幾個獨立的 Crew 組合成一條有邏輯的流程

最小可用的 Flow 範例

from pydantic import BaseModel
from crewai.flow.flow import Flow, start, listen, router
 
class BlogState(BaseModel):
    topic: str = ""
    draft: str = ""
    word_count: int = 0
    approved: bool = False
 
class BlogFlow(Flow[BlogState]):
 
    @start()
    def init_topic(self) -> str:
        self.state.topic = "CrewAI for beginners"
        return self.state.topic
 
    @listen(init_topic)
    def write_draft(self, topic: str) -> str:
        # 這裡可以呼叫一個 WritingCrew
        draft = f"Draft content about {topic}..."
        self.state.draft = draft
        self.state.word_count = len(draft.split())
        return draft
 
    @router(write_draft)
    def quality_gate(self) -> str:
        if self.state.word_count > 100:
            return "approved"
        return "retry"
 
    @listen("approved")
    def publish(self) -> str:
        return f"Published: {self.state.draft[:50]}..."
 
    @listen("retry")
    def revise(self) -> str:
        return "Draft too short, revising..."

@start() 是流程入口，@listen() 監聽前一步的完成，@router() 根據條件決定走哪條分支。

Flow[BlogState] 裡的 BlogState 是整個流程的共享狀態——每個方法都可以讀取和修改它，這樣資料就可以在步驟之間流動，不需要全域變數。

狀態管理：Flow 最重要的設計概念

Flow 的狀態是整個架構的核心。把它用 Pydantic BaseModel 定義，有幾個好處：

有型別提示，IDE 能幫你抓錯
所有方法共享同一份狀態，資料流動清楚
容易序列化和儲存，方便 debug

不要用裸 dict 當狀態——你半年後回來看，會花很多時間猜每個 key 是什麼意思。

什麼時候才需要用 Flow？

不是所有情境都需要 Flow。判斷標準很簡單：

流程超過 3 個步驟，且步驟之間有條件判斷 → 考慮 Flow
需要多個 Crew 協作，且執行順序不是全部都跑 → 考慮 Flow
有「重試」或「審核不通過回頭重做」的邏輯 → 一定要用 Flow

如果你只是「研究員找資料 + 編輯寫報告」，繼續用單一 Crew 就好，不需要 Flow。引入 Flow 的成本是增加程式碼複雜度，只有在你真的需要那些能力的時候才值得。

Flow 的 3 個設計原則

一個方法一件事：每個 @listen 方法只負責一個步驟。如果一個方法在做三件事，把它拆開——Flow 的可讀性和 debug 難度都會變好。

條件邏輯集中在 @router：不要把 if/else 散落在各個 @listen 方法裡。@router 就是設計來集中管理分支邏輯的，用它。

狀態用 BaseModel 定型：裸 dict 最省事，但你一個月後就不知道自己在幹嘛了。

常見問題

Q：Flow 和 Crew 可以混用嗎？
可以，這正是 Flow 的設計用途。在 @listen 方法裡建立並執行 Crew，把 Crew 的輸出存到 self.state，後續步驟再讀取。

Q：@router 的回傳值有什麼限制？
回傳值是字串，會對應到 @listen 裡監聽的字串。確保 router 的每個可能回傳值都有對應的 listener，不然流程會在那個分支卡死。

Q：Flow 可以暫停執行等待人工確認嗎？
可以，透過 pause_for_input 機制。這個功能適合需要「人工審核後才繼續」的業務流程，但需要搭配持久化狀態使用，設定稍微複雜一點。

下一步

流程能跑了，但你可能注意到一件事：Agent 每次執行都是全新的，不記得上次說了什麼。下一篇來解決這個問題：
👉 Memory 與 Knowledge