#crewai#flow#workflow#state-machine#python

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 1 メソッド = 1 役割
  2. 分岐ロジックは @router に寄せる
  3. 状態は BaseModel で型定義し、裸 dict を避ける

この形なら、半年後に見返しても発掘作業になりません。

いつ Flow を使うべきか

  • 3 ステップを超えるフロー
  • 明確な分岐条件がある
  • 複数 Crew の協調が必要

「調査 + レポート」程度なら単一 Crew で十分です。

次へ

フローが動くと次に出る疑問は「文脈を覚えてくれるか？」です。
次回は Memory と Knowledge source を扱います。
👉 Memory と Knowledge