15. Flows: 이벤트 기반 프로덕션 오케스트레이션

비유로 시작하기: 자율 팀 vs 공정 관리자

14장의 Crew는 자율적인 팀입니다. 직원들에게 목표를 주면 알아서 협력해 결과를 냅니다. 자율성이 높은 대신, "정확히 이 조건일 때 이 단계로 가라" 같은 세밀한 통제는 어렵습니다.

Flow는 그 위에 얹는 공정 관리자입니다. "언제 작업이 시작되고, 어떤 상태가 다음으로 넘어가고, 조건에 따라 어디로 분기하는지"를 정밀하게 통제합니다. 그리고 필요할 때 Flow 안에서 Crew를 호출해 자율 작업을 시킵니다. 즉 Flow는 통제, Crew는 자율입니다.

CrewAI 공식 문서도 프로덕션 앱은 Flow로 구조를 잡고 그 안에서 Crew를 쓰라고 권합니다. "모든 문제에 에이전트 떼를 던지지 말고, 통제된 워크플로를 먼저 만들고 그 안에서 에이전트를 쓰라"는 철학입니다.

Crews vs Flows 한눈에

flowchart TB
    classDef flow fill:#FFE082,stroke:#F9A825,color:#000
    classDef crew fill:#A5D6A7,stroke:#2E7D32,color:#000
    classDef llm fill:#80DEEA,stroke:#00838F,color:#000

    subgraph FLOW[Flow · 통제·이벤트 기반]
    S["@start 진입점"]:::flow --> L["@listen 단계"]:::flow
    L --> R{"@router 분기"}:::flow
    R -->|조건 A| C[Crew 호출<br/>자율 협업]:::crew
    R -->|조건 B| LL[단일 LLM 호출]:::llm
    end

Flow의 세 데코레이터

Flow는 파이썬 클래스이고, 메서드에 데코레이터를 붙여 흐름을 정의합니다. 핵심은 세 가지입니다.

@start()는 흐름의 진입점을 표시합니다. @listen(...)은 지정한 메서드가 끝나면 이어서 실행될 리스너를 표시합니다. @router(...)는 계산된 조건에 따라 분기합니다.

from crewai.flow.flow import Flow, start, listen, router
from pydantic import BaseModel

# Flow가 단계 간에 공유할 상태 (Pydantic 모델)
class ContentState(BaseModel):
    topic: str = ""
    is_technical: bool = False
    draft: str = ""

class ContentFlow(Flow[ContentState]):

    @start()
    def set_topic(self):
        self.state.topic = "2026 AI 에이전트 동향"
        # 간단한 판정 (실전에선 LLM 사용 가능)
        self.state.is_technical = "에이전트" in self.state.topic

    @router(set_topic)
    def decide_path(self):
        # 다음에 실행할 리스너의 이름(라벨)을 반환
        return "technical" if self.state.is_technical else "general"

    @listen("technical")
    def write_technical(self):
        self.state.draft = f"[기술 심화] {self.state.topic} 보고서"
        return self.state.draft

    @listen("general")
    def write_general(self):
        self.state.draft = f"[일반] {self.state.topic} 소개"
        return self.state.draft

flow = ContentFlow()
result = flow.kickoff()
print(result)

흐름은 이렇습니다. set_topic이 진입점으로 실행되어 상태를 채우고, @router가 상태를 보고 "technical" 또는 "general"을 반환하며, 해당 라벨을 듣는 @listen 메서드가 실행됩니다. 12장 LangGraph의 조건부 엣지와 매우 닮은 분기 구조입니다.

📌 핵심: Flow의 self.state는 LangGraph의 State와 같은 역할입니다. 단계 간 데이터를 실어 나르는 공유 메모리입니다. Pydantic 모델로 정의하면 타입 검증까지 됩니다.

Flow 안에서 Crew 호출하기

Flow의 진가는 단계 중 하나에서 Crew를 통째로 실행할 때 나옵니다. 통제는 Flow가, 자율 작업은 Crew가 맡는 분업입니다.

from crewai.flow.flow import Flow, start, listen

class ResearchFlow(Flow):
    @start()
    def prepare(self):
        return "AI 에이전트 시장"

    @listen(prepare)
    def run_crew(self, topic):
        # 14장에서 만든 크루를 여기서 호출
        result = crew.kickoff(inputs={"topic": topic})
        return result

Flow가 입력을 준비하고(prepare), 그 결과를 받아 Crew를 가동합니다(run_crew). Flow는 언제·무엇을·어떤 순서로 할지 통제하고, Crew는 그 안에서 자율적으로 협업합니다.

⚠️ 흔한 실수: 단순한 순차 작업에 Flow를 과하게 쓰는 것. 단계가 정해져 있고 분기도 없다면 Crew의 Sequential만으로 충분합니다. 조건 분기·상태 관리·여러 크루 조합이 필요할 때 Flow를 도입하세요.

💡 팁: Flow는 or_, and_ 같은 논리 연산자로 복합 조건을 만들 수 있습니다. "A 또는 B가 끝나면 실행" 같은 트리거를 구성할 때 유용합니다.

언제 Crew, 언제 Flow

flowchart TD
    classDef q fill:#FFE082,stroke:#F9A825,color:#000
    classDef crew fill:#A5D6A7,stroke:#2E7D32,color:#000
    classDef flow fill:#90CAF9,stroke:#1565C0,color:#000

    Q1{조건 분기·상태 관리·<br/>여러 크루 조합이 필요한가?}:::q
    Q1 -->|아니오| C[Crew만으로 충분<br/>Sequential/Hierarchical]:::crew
    Q1 -->|예| F[Flow로 통제<br/>+ 내부에서 Crew 호출]:::flow

💡 실습 아이디어(강의의 데이터사이언스 워크플로 대응): 다음 장에서 만들 ML 워크플로를 Flow로 감싸, "데이터 검증 통과 → 모델 학습 크루 실행 → 평가 점수가 낮으면 재학습 분기" 같은 통제 구조를 얹어 보세요.

이 장에서 배운 것

• Crew는 자율적 협업, Flow는 통제된 이벤트 기반 흐름이다. Flow 안에서 Crew를 호출해 둘을 결합한다.
• @start(진입점)·@listen(이어 실행)·@router(분기) 세 데코레이터로 흐름을 정의한다.
• self.state(Pydantic)가 단계 간 데이터를 나르는 공유 메모리다. LangGraph의 State와 유사하다.
• 단순 순차 작업엔 Crew, 조건 분기·상태·다중 크루엔 Flow를 쓴다.

✍️ 확인 문제

1. "Flow는 통제, Crew는 자율"이라는 문장을 콘텐츠 제작 파이프라인 예로 풀어 설명하라.
2. @router가 반환한 값은 무엇과 연결되어 다음 실행을 결정하는가?
3. 단계가 고정된 2단계 작업에 Flow를 쓰는 것이 과한 이유는 무엇인가?

이전 장: 14. CrewAI 입문
다음 장: 16. 실전: 데이터사이언스 크루