CrewAI Flows — 워크플로우를 코드로 엮다 (Flows)

여러 Crew와 태스크를 단순히 하나씩 돌리는 걸로는 끝나지 않아요. 어떤 결과가 나왔는지 보고 다음 단계를 정하고, 중간 상태를 챙기고, 사람의 판단을 끼워 넣어야 하는 순간이 옵니다. CrewAI Flows는 바로 그런 "여러 단계를 이어 붙이는 일"을 구조화된, 이벤트 중심(event-driven) 방식으로 풀어 주는 기능이에요. 쉽게 말해 Crew라는 부품 여러 개를 하나의 파이프라인처럼 엮어 주는 프레임워크라고 보면 돼요.

Flows가 주는 핵심 이점은 네 가지로 정리돼요.

1. 워크플로우를 쉽게 만들기 — 여러 Crew와 태스크를 이어 붙여 복잡한 AI 워크플로우를 구성할 수 있어요.
2. 상태 관리(state management) — 워크플로우의 여러 단계 사이에서 데이터를 주고받고 공유하기가 매우 쉬워요.
3. 이벤트 중심 구조 — 어떤 작업이 "출력을 냈을 때" 다음 작업이 반응하도록 이벤트 모델 위에서 움직여요.
4. 유연한 제어 흐름 — 조건 분기, 반복, 여러 갈래로 나누는 흐름을 코드로 직접 제어할 수 있어요.

시작하기

가장 간단한 예부터 볼게요. OpenAI로 첫 번째 작업에서 랜덤한 도시 이름을 하나 만들고, 두 번째 작업에서 그 도시에 대한 재미있는 사실을 생성하는 Flow예요.

from crewai.flow import Flow, start, listen

class CityFunFactFlow(Flow):
    @start()
    def generate_city(self):
        import random
        cities = ["Paris", "Tokyo", "Seoul", "New York"]
        city = random.choice(cities)
        return city

    @listen(generate_city)
    def generate_fun_fact(self, city):
        return f"Did you know {city} has a great story?"

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

이 Flow는 generate_city라는 태스크에서 시작해서, 그 출력을 generate_fun_fact가 "듣고(listen)" 이어받는 구조예요. 여기서 주목할 점이 몇 가지 있어요.

• Flow 인스턴스는 실행될 때 자동으로 고유한 식별자(UUID) 를 상태(state)에 부여받아요. 실행을 추적하고 관리하는 데 쓰여요.
• 상태에는 생성된 도시나 재미있는 사실 같은 추가 데이터도 저장돼요. 흐름이 진행되는 동안 계속 유지돼요.
• 실행 순서를 정리하면 — 상태용 고유 ID 생성 → 랜덤 도시 생성 후 저장 → 그 도시에 대한 재미있는 사실 생성 후 저장 → 결과를 콘솔에 출력, 이렇게 돼요.
• OpenAI를 호출한다면 .env 파일에 OPENAI_API_KEY를 반드시 설정해 둬야 해요. 이 키가 OpenAI API 인증에 필요하니까요.

@start() — 흐름의 시작점

@start() 데코레이터는 Flow의 진입점(entry point) 을 표시해요. 여러 방식으로 쓸 수 있어요.

• 조건 없이 시작점을 여러 개 선언할 수 있어요: @start() — Flow가 시작되거나 재개될 때 만족된 모든 @start() 메서드가 실행돼요(보통 병렬로요).
• 특정 메서드나 라우터 레이블 뒤에서 시작하도록 문을 걸 수 있어요: @start("method_or_label").
• 호출 가능한 조건(callable condition)을 넘겨서 시작 시점을 제어할 수도 있어요.

@listen() — 출력을 듣는 리스너

@listen() 데코레이터는 어떤 태스크의 출력에 반응하는 메서드를 표시해요. 지정한 태스크가 출력을 냈을 때 실행되는데, 그 태스크의 출력을 인자로 받을 수 있어요.

@listen()은 여러 방식으로 쓸 수 있어요.

1. 메서드 이름으로 듣기 — 문자열로 메서드 이름을 넘겨요. 그 메서드가 끝나면 리스너가 실행돼요.
2. 메서드를 직접 넘겨 듣기 — 메서드 자체를 넘겨요. 그것이 완료되면 리스너가 실행돼요.

@listen("generate_city")      # 이름으로 듣기
def step_a(self, city): ...

@listen(generate_city)        # 메서드 자체로 듣기
def step_a(self, city): ...

Flow 출력 다루기

Flow의 최종 출력은 마지막으로 완료된 메서드의 출력으로 결정돼요. kickoff()를 호출하면 그 최종 메서드의 출력이 반환돼요. 그리고 plot() 메서드를 쓰면 흐름을 시각화한 HTML 파일이 생성돼요.

result = flow.kickoff()
print(result)
flow.plot()   # my_flow_plot.html 생성

출력 이후에도 상태(state)에 접근할 수 있어요. 실행 중 각 메서드가 상태에 추가하거나 수정한 모든 정보를 실행 후에 꺼내 볼 수 있어요. 즉, 최종 결과 하나만 받는 게 아니라 흐름 전체가 겪은 과정의 상태까지 들여다볼 수 있는 구조예요.

Flow 사용 지표 (usage_metrics)

Flow 실행이 끝나면 usage_metrics 속성으로 실행 중 발생한 모든 LLM 호출의 토큰 사용량 합계를 확인할 수 있어요. Flow가 조율한 각 Crew의 호출, Agent 내부 툴이 쓰는 호출, Flow 메서드 안에서 직접 부른 LLM.call(...)까지 모두 집계돼요. 이 값은 CrewAI Enterprise UI에 표시되는 합계와 동일한 수치의 SDK 쪽 버전이라고 보면 돼요.

• 반환되는 각 항목(UsageMetrics)은 단일 flow.kickoff() 안에서 발생한 모든 LLM 호출의 합이에요.
• 카운터는 다음 kickoff() 호출 때(또는 kickoff_for_each의 각 반복마다) 리셋돼요. 그래서 연속 실행해도 중복 집계되지는 않아요.
• kickoff()가 끝난 뒤에는 아무 때나 읽어도 안전해요. 실행 중에 읽으면 그때까지 누적된 부분 합계가 나와요.

Flow 상태 관리

신뢰할 수 있는 워크플로우를 만들려면 상태를 잘 관리하는 게 핵심이에요. CrewAI Flows는 비구조화(unstructured) 와 구조화(structured) 두 방식의 상태 관리를 모두 제공해서, 애플리케이션에 맞는 쪽을 고를 수 있어요.

비구조화 상태 관리

모든 상태를 Flow 클래스의 state 속성에 저장하는 방식이에요. 엄격한 스키마(데이터 형식 정의) 없이 실행 중에 상태 속성을 자유롭게 추가하거나 바꿀 수 있는 게 장점이에요. 비구조화 상태에서도 CrewAI는 각 상태 인스턴스에 고유 식별자(UUID)를 자동 생성하고 유지해요.

from crewai.flow import Flow

class MyFlow(Flow):
    @start()
    def method_a(self):
        self.state.message = "hello"   # 스키마 없이 자유롭게 추가

flow = MyFlow()
flow.kickoff()
print(flow.state.message)

• id 필드는 자동 생성되어 흐름 실행 내내 유지돼요. 직접 관리하거나 설정할 필요 없고, 새 데이터로 상태를 갱신해도 그대로 유지돼요.
• 유연성: 미리 정해진 제약 없이 self.state에 속성을 동적으로 추가할 수 있어요.
• 단순함: 상태 구조가 단순하거나 자주 바뀌는 직관적인 워크플로우에 잘 맞아요.

구조화 상태 관리

미리 정의한 스키마를 활용해서 워크플로우 전반의 일관성과 타입 안전성을 보장하는 방식이에요. Pydantic의 BaseModel 같은 모델로 상태의 정확한 형태를 정의해 두면, 검증과 개발 환경의 자동 완성(auto-completion)을 더 잘 활용할 수 있어요.

from pydantic import BaseModel
from crewai.flow import Flow

class ExampleState(BaseModel):
    counter: int = 0
    message: str = ""

class MyFlow(Flow):
    @start()
    def method_a(self):
        self.state.message = "hello"

flow = MyFlow()
flow.kickoff()
print(flow.state.message)

구조화 상태의 장점을 정리하면 —

• 명확한 스키마: 상태 구조가 코드에 드러나서 읽기와 유지보수가 좋아져요.
• 타입 안전성: Pydantic 덕분에 상태 속성이 지정된 타입을 따르게 되어 런타임 오류가 줄어요.
• 자동 완성: IDE가 상태 모델에 맞춰 더 나은 자동 완성과 오류 검사를 제공해요.

무엇을 쓸까?

• 비구조화: 상태가 단순하거나 매우 동적일 때, 엄격한 정의보다 유연성이 우선일 때, 스키마 정의 부담 없이 빠르게 프로토타입을 만들고 싶을 때.
• 구조화: 잘 정의되고 일관된 상태 구조가 필요할 때, 타입 안전성과 검증이 안정성에 중요할 때, IDE의 자동 완성·타입 검사를 개발 경험에 활용하고 싶을 때.

요약하면, 유연함이 필요한가, 안정성이 필요한가에 따라 고르면 돼요.

Flow 영속화 (Persistence) — @persist

@persist 데코레이터는 상태 영속화를 자동으로 켜 주는 역할이에요. 덕분에 서버가 재시작되거나 다른 실행으로 넘어가도 흐름의 상태를 유지할 수 있어요. 클래스 레벨이나 메서드 레벨 어디에 붙이느냐에 따라 유연하게 제어할 수 있어요.

• 클래스 레벨: 클래스에 붙이면 모든 Flow 메서드의 상태를 자동으로 영속화해요.
• 메서드 레벨: 개별 메서드에만 붙여 더 세밀하게 영속화 대상을 골라요.

@persist는 kickoff / kickoff_async에서 두 가지 수화(hydration) 방식을 지원해요.

• kickoff(inputs={"id": <uuid>}) — 이어서(resume): 주어진 UUID의 최신 스냅샷을 불러와 같은 flow_uuid 아래에서 계속 써 내려가요. 이력이 이어져요.
• kickoff(restore_from_state_id=<uuid>) — 분기(fork): 주어진 UUID의 최신 스냅샷을 불러와 새 실행의 상태를 그것으로 채우되, 새로운 state.id를 부여해요(자동 생성되거나, inputs["id"]로 고정). 새 실행의 @persist 기록은 새 state.id 아래에 쌓이고, 원본 흐름의 이력은 그대로 보존돼요.

어떻게 동작하나요?

1. 고유 상태 식별 — 각 Flow 상태는 자동으로 UUID를 받아요. 상태 갱신과 메서드 호출을 거쳐도 유지되고, 구조화(Pydantic)와 비구조화(딕셔너리) 상태 모두를 지원해요.
2. 기본 SQLite 백엔드 — SQLiteFlowPersistence가 기본 저장소예요. 상태가 자동으로 로컬 SQLite 데이터베이스에 저장돼요. 데이터베이스 작업이 실패하면 명확한 에러 메시지로 알려줘요.
3. 오류 처리 — 데이터베이스 작업에 대한 자세한 에러 메시지, 저장·불러오기 중 자동 상태 검증, 영속화 문제가 발생했을 때의 명확한 피드백을 제공해요.

알아둘 점

• 상태 타입: 구조화(Pydantic BaseModel)와 비구조화(딕셔너리) 상태 모두 지원해요.
• 자동 ID: id 필드는 없으면 자동으로 추가돼요.
• 상태 복구: 실패했거나 재시작된 Flow는 이전 상태를 자동으로 다시 불러올 수 있어요.
• 커스텀 구현: 특수한 저장 요구가 있다면 직접 FlowPersistence 구현체를 제공할 수도 있어요.

restore_from_state_id가 일치하는 영속화된 상태가 없으면 기존 inputs["id"] resume의 not-found 동작처럼 조용히 기본 동작으로 넘어가요. restore_from_state_id와 from_checkpoint를 함께 쓰면 ValueError가 나요 — 수화 소스는 하나만 고르세요. 사람이 분기(fork)할 때 inputs["id"]를 고정하면 다른 Flow와 영속화 키를 공유하게 되니, 보통은 restore_from_state_id만 쓰는 게 안전해요.

흐름 제어 (Flow Control)

단순히 이어 붙이는 걸 넘어, 조건과 갈래를 다룰 수 있다는 게 Flow의 강점이에요.

or 조건 — 여러 출력 중 하나

or_ 함수를 쓰면 여러 메서드를 동시에 "듣고", 그중 어느 하나라도 출력을 내면 리스너를 실행해요.

from crewai.flow import Flow, listen, or_

@listen(or_(start_method, second_method))
def logger(self, event): ...

이 Flow를 돌리면 logger는 start_method나 second_method 중 하나의 출력에 의해 실행돼요.

and 조건 — 모두 출력해야

and_ 함수는 여러 메서드를 듣되, 모든 메서드가 출력을 냈을 때만 리스너를 실행해요.

from crewai.flow import Flow, listen, and_

@listen(and_(start_method, second_method))
def logger(self, event): ...

Router — 출력값에 따라 갈래 나누기

@router() 데코레이터는 메서드 출력값을 기준으로 조건부 분기를 정의하게 해줘요. 어떤 메서드의 결과에 따라 서로 다른 경로로 흐름을 동적으로 제어할 수 있어요.

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

@router(start_method)
def second_method(self):
    return "success" if self.state.flag else "failed"

@listen("success")
def third_method(self): ...

@listen("failed")
def fourth_method(self): ...

예시로, start_method가 랜덤한 불리언 값을 만들어 상태에 넣고, second_method가 그 값을 보고 "success" 또는 "failed"를 반환하면, third_method와 fourth_method가 각자 해당 라벨을 듣고 실행돼요.

사람의 개입 (Human in the Loop)

@human_feedback 데코레이터는 인간의 피드백을 모으기 위해 흐름을 일시 중지시켜요. 승인 게이트(approval gate), 품질 검토, 인간의 판단이 필요한 의사 결정 지점에 유용해요.

from crewai.flow import Flow, listen, human_feedback

@human_feedback
def collect_feedback(self): ...

• emit을 지정하면, 사람의 자유 형식 피드백을 LLM이 해석해서 지정된 결과 중 하나로 좁혀 주고, 그 결과에 해당하는 @listen을 트리거해요.
• 라우팅 없이 단순히 피드백만 수집하려면 @human_feedback을 그대로 쓰면 돼요.
• 실행 중 모인 피드백은 self.last_human_feedback(가장 최근)이나 self.human_feedback_history(전체 목록)로 접근할 수 있어요.
• 비동기·비차단(non-blocking) 피드백을 Slack, 웹훅 같은 커스텀 프로바이더로 받는 것까지 자세히 보려면 "Human Feedback in Flows" 문서를 확인하세요. (확인 필요 — 자세한 절차는 별도 가이드 참조)

Flow에 Agent 넣기

항상 완전한 Crew가 필요한 건 아니에요. 더 단순하고 집중된 태스크 실행이 필요할 땐 Flow에 Agent 하나를 직접 넣을 수도 있어요. 예를 들어 시장 조사를 하는 Agent를 Flow 안에 넣는 식이에요. 이 패턴의 핵심 포인트는 —

1. 구조화 출력(structured output): Pydantic 모델로 출력 형식(MarketAnalysis)을 정의해 두면 타입 안전성과 구조화된 데이터가 흐름 전반을 지켜줘요.
2. 상태 관리: Flow 상태(MarketResearchState)가 단계 사이의 맥락을 유지하고 입력과 출력을 모두 보관해요.
3. 툴 통합: Agent는 WebsiteSearchTool 같은 툴을 이용해 능력을 확장할 수 있어요.

Flow에 Crew 넣기

crewai create flow name_of_flow 명령으로 새 CrewAI 프로젝트를 만들면, 여러 Crew를 포함하는 Flow를 만들 수 있는 뼈대(scaffolding)가 생성돼요. 생성된 프로젝트에는 이미 동작하는 poem_crew라는 사전 구축 크루가 포함돼요. 임베디드 스타터 크루는 고전적인 Python/YAML 구조를 쓰고, crewai create crew로 새로 만든 독립 크루는 JSON-first 구조를 쓴다는 차이가 있어요.

생성 후 폴더 구조는 대략 —

• crews/ 폴더에 여러 크루를 정의할 수 있어요.
• 생성된 poem_crew는 고전적인 임베디드 크루 구조를 써요:
• config/agents.yaml — 크루의 에이전트 정의
• config/tasks.yaml — 크루의 태스크 정의
• poem_crew.py — 에이전트·태스크·크루 자체를 담은 크루 정의

poem_crew를 복사·붙여넣기·수정해서 다른 고전 임베디드 크루를 만들 수 있어요. JSON-first 임베디드 크루는 crew.jsonc와 agents/*.jsonc가 있는 폴더를 쓰고, Flow 단계에서 불러와 사용해요.

main.py에서 Flow 클래스와 @start, @listen 데코레이터로 흐름을 정의하고 여러 크루를 이어 붙여요. 예시에서는 문장 수를 세고, PoemCrew로 시를 만들고, 그 시를 파일에 저장하는 PoemFlow를 만들어 kickoff()로 실행하는 구조예요.

실행하기

의존성 설치(선택), 가상 환경 활성화 후 CrewAI CLI로 흐름을 실행하거나 프로젝트 스크립트를 직접 실행할 수 있어요. 실행하면 콘솔에 결과가 보여요.

Flow 실행하기

Flow를 실행하는 방법은 두 가지가 있어요.

Flow API로 실행

Flow 클래스의 인스턴스를 만든 뒤 kickoff() 메서드를 호출하는 방식이에요.

flow = MyFlow()
result = flow.kickoff()

스트리밍으로 실행

실시간으로 실행 흐름을 보면서 출력이 만들어지는 대로 받아보고 싶다면 스트리밍을 켜면 돼요. 스트리밍 흐름 실행에 대한 자세한 내용은 "Streaming Flow Execution" 가이드를 참고하세요. (확인 필요)

CLI로 실행하기

버전 0.103.0부터는 crewai run 명령으로 Flow를 실행할 수 있어요.

crewai run

이 명령은 pyproject.toml의 type = "flow" 설정을 보고 프로젝트가 Flow인지 자동으로 판단해 그에 맞게 실행해요. CLI에서 Flow를 실행하는 권장 방법이에요. 예전의 crewai flow kickoff 명령은 더 이상 사용되지 않습니다(deprecated). 크루와 Flow 모두 crewai run을 쓰세요.

Flow의 메모리

모든 Flow는 CrewAI의 통합 메모리(Memory) 시스템에 자동으로 접근할 수 있어요. 기본 Memory() 인스턴스는 Flow가 초기화될 때 자동 생성되고, 필요하면 커스텀 인스턴스를 넘길 수도 있어요. 어느 Flow 메서드에서나 저장·회상·추출을 위한 세 가지 내장 편의 메서드를 쓸 수 있어요.

메모리는 디스크의 LanceDB에 저장되어 실행 간에도 유지되기 때문에, analyze 같은 단계가 이전 실행에서 배운 내용까지 회상할 수 있어요. 즉 흐름이 시간이 지나며 학습하고 지식을 축적하는 게 가능해져요. 메모리의 스코프, 슬라이스, 복합 점수, 임베더 설정 등 자세한 내용은 "Memory" 문서를 참고하세요. (확인 필요)

Plot — 흐름 시각화

워크플로우를 그림으로 보면 구조와 실행 경로를 이해하기 쉬워져요. CrewAI는 Flow의 인터랙티브 플롯(plot) 을 생성하는 시각화 도구를 제공해요. 태스크, 연결 관계, 데이터 흐름을 그래픽으로 보여주기 때문에 실행 순서를 파악하고, 병목을 찾고, 워크플로우 로직이 의도와 맞는지 확인하는 데 도움이 돼요.

두 가지 방법으로 플롯을 만들 수 있어요.

1. plot() 메서드: Flow 인스턴스에서 flow.plot()을 호출하면 현재 디렉터리에 my_flow_plot.html이라는 인터랙티브 플롯 HTML 파일이 생성돼요. 브라우저로 열어 보면 돼요.
2. 커맨드 라인 사용: CLI로도 플롯을 생성할 수 있어요. (확인 필요)

실제 활용 예시

CrewAI 공식 예시 리포지토리에서 이런 응용을 볼 수 있어요.

1. Email Auto Responder Flow — 백그라운드 작업이 계속 돌며 이메일 응답을 자동화하는 무한 루프 구조. 반복 개입이 필요 없는 태스크에 아주 좋아요.
2. Lead Score Flow — human-in-the-loop(인간 피드백)와 라우터를 이용한 조건 분기를 보여주는 예시. 동적 의사 결정과 사람의 검토를 워크플로우에 섞는 방법을 다뤄요.
3. Write a Book Flow — 여러 크루를 이어 붙이는 예시. 한 크루가 책 전체의 개요(outline)를 만들고 다른 크루가 그 개요를 바탕으로 장(chapter)을 생성해서, 최종적으로 완결된 책 하나를 만들어요.
4. Meeting Assistant Flow — 한 이벤트를 여러 후속 작업에 방송(broadcast)하는 예시. 회의가 끝나면 Trello 보드를 갱신하고 Slack 메시지를 보내고 결과를 저장하는 식이에요. 단일 이벤트에서 여러 결과를 처리하는 데 좋아요.

반복적인 태스크 자동화부터 인간의 피드백이 필요한 복잡한 다단계 프로세스까지, CrewAI Flows를 어떤 식으로 활용할 수 있는지 감을 잡기 좋은 예시들이에요.

데이터스케쳐스 실무 관점

데이터 프로젝트에서 CrewAI Flows가 특히 반가운 지점은 "분석 파이프라인을 코드 하나로 정의하고, 중간 결과를 손에 쥔 채로 둘 수 있다"는 부분이에요. 실제 ETL이나 리포트 생성 워크플로우를 만들다 보면 단계마다 결과가 다음 입력이 되는 체인이 많은데, @start()/@listen() 구조는 그 체인을 함수 호출처럼 붙일 수 있게 해줘요. 무엇보다 상태(state)를 명시적으로 관리할 수 있다는 점이 실무에서 큰 차이를 만들어요 — 민감하거나 추적이 필요한 값은 Pydantic BaseModel로 구조화 상태를 정의해 타입 안전성을 챙기고, 실험 단계에서 빠르게 이것저것 시도해 볼 땐 비구조화 상태로 유연하게 넘어갈 수 있어요.

다만 두 가지를 짚어야 해요. 첫째, LLM 호출 비용이 신경 쓰이는 운영 환경이라면 usage_metrics가 단일 kickoff() 단위로 토큰 합계를 낸다는 점을 기억하고, 실행 단위를 설계할 때 이걸 반영하는 게 좋아요. 둘째, 사람의 검토가 필요한 지점(수치 검증, 문안 승인 같은 곳)에는 @human_feedback이나 @router를 끼워 넣어 자동화와 사람 판단 사이의 균형을 잡을 수 있어요. 요청이 길거나 결제가 얽힌 파이프라인이라면 @persist로 상태를 영속화해 중간에 끊겨도 재개될 수 있게 만드는 것도 실전에서 높은 가치를 줍니다. Crew보다 가벼운 단일 에이전트 처리가 필요할 땐 굳이 Crew를 만들지 않고 Agent를 Flow에 직접 넣는 선택지도 꼭 염두에 두세요.

더 알아보기

• 공식 문서: CrewAI Flows — docs.crewai.com/edge/en/concepts/flows
• 스트리밍 가이드: Streaming Flow Execution (확인 필요 — 원문 경로 기준)
• 사람 피드백 가이드: Human Feedback in Flows (확인 필요 — 원문 경로 기준)
• 메모리 문서: CrewAI Memory — /concepts/memory (확인 필요 — 원문 경로 기준)