CrewAI Flow 入門:把多個 Crew 串成可控流程

2 min read

當你的流程變成「研究 -> 產文 -> 審核 -> 發布」,單一 Crew 就不夠了。
這時候 Flow 就像總控台,負責協調多個 Crew 的執行順序。

Flow 在解什麼問題?

  • 管理多步驟狀態
  • 做條件分支(通過就發布,不通過就重跑)
  • 讓流程更可觀測、可維護

最小 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
 
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:
        result = "draft content..."
        self.state.draft = result
        self.state.word_count = len(result.split())
        return result
 
    @router(write_draft)
    def quality_gate(self) -> str:
        return "ok" if self.state.word_count > 100 else "retry"

Flow 設計建議

  1. 一個 @listen 方法只做一件事
  2. 分支邏輯放在 @router,不要塞滿 if/else
  3. 狀態用 BaseModel 明確型別,不要裸 dict

這樣你半年後回來看,不會像在考古。

什麼時候該用 Flow?

  • 一個流程超過 3 個步驟
  • 有明確分支條件
  • 需要多個 Crew 協作

如果你只是「研究 + 寫報告」,先用單 Crew 即可。

下一步

流程能跑之後,大家最常問:「Agent 會記得前文嗎?」
下一篇就來處理記憶與知識來源:
👉 Memory 與 Knowledge