CrewAI Flows

여러 Crew와 태스크를 이어 붙여 복잡한 AI 워크플로를 만드는 일, CrewAI에서는 Flows가 그 역할을 맡아요. Flows는 각 단계가 이벤트로 연결되는 구조라서, 여러 Crew를 사슬처럼 엮고 단계 사이의 상태를 주고받으며 조건 분기나 반복까지 자유롭게 제어할 수 있어요. 이 페이지에서 Flows의 기본 개념부터 상태 관리, 조건 로직, 플로우에 Agent·Crew를 얹는 법까지 한 번에 정리합니다.

출처: 공식문서

본문

Flows가 해결하는 것

CrewAI Flows는 AI 자동화를 만들 때 다음 네 가지를 쉽게 해 줍니다.

워크플로 생성 단순화 — 여러 Crew와 태스크를 사슬처럼 연결해 복잡한 AI 워크플로를 구성합니다.
상태 관리 — 워크플로의 서로 다른 단계 사이에 상태를 저장·공유하기 좋습니다.
이벤트 기반 구조 — 이벤트를 기반으로 동작해 동적이고 반응성 있는 워크플로를 만들 수 있어요.
유연한 제어 흐름 — 조건 로직, 루프, 분기를 워크플로 안에서 구현할 수 있습니다.

시작하기: 랜덤 도시로 재미있는 사실 만들기

가장 단순한 예시부터 볼게요. 한 태스크에서 OpenAI로 랜덤 도시를 만들고, 그 결과를 받아 다른 태스크에서 재미있는 사실을 생성하는 Flow입니다.

from crewai.flow.flow import Flow, listen, start
from dotenv import load_dotenv
from litellm import completion

load_dotenv()

class ExampleFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def generate_city(self):
        print("Starting flow")
        # Each flow state automatically gets a unique ID
        print(f"Flow State ID: {self.state['id']}")

        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": "Return the name of a random city in the world.",
                },
            ],
        )

        random_city = response["choices"][0]["message"]["content"]
        # Store the city in our state
        self.state["city"] = random_city
        print(f"Random City: {random_city}")

        return random_city

    @listen(generate_city)
    def generate_fun_fact(self, random_city):
        response = completion(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": f"Tell me a fun fact about {random_city}",
                },
            ],
        )

        fun_fact = response["choices"][0]["message"]["content"]
        # Store the fun fact in our state
        self.state["fun_fact"] = fun_fact
        return fun_fact



flow = ExampleFlow()
flow.plot()
result = flow.kickoff()

print(f"Generated fun fact: {result}")

여기서 generate_city는 Flow의 시작점이고, generate_fun_fact는 @listen(generate_city)로 앞 단계의 출력을 기다렸다가 실행돼요. 각 Flow 인스턴스는 상태에 자동으로 고유한 UUID를 받고, 생성한 도시와 재미있는 사실 같은 추가 데이터도 상태에 담아 Flow 실행 동안 유지합니다.

실행을 돌리면 Flow가 이런 순서로 동작해요.

Flow 상태용 고유 ID 생성
랜덤 도시를 생성해 상태에 저장
그 도시에 대한 재미있는 사실을 생성해 상태에 저장
결과를 콘솔에 출력

참고: .env 파일에 OPENAI_API_KEY를 설정해 두어야 OpenAI API 요청이 인증됩니다.

`@start()` — Flow의 시작점

@start() 데코레이터는 Flow의 진입점을 표시합니다.

무조건 시작하는 진입점을 여러 개 선언할 수 있어요: @start()
앞선 메서드나 라우터 라벨에 시작을 게이트할 수 있어요: @start("method_or_label")
호출 가능한 조건을 넘겨 시작 시점을 제어할 수도 있습니다.

만족되는 모든 @start() 메서드는 Flow가 시작(또는 재개)될 때 (종종 병렬로) 실행됩니다.

`@listen()` — 다른 단계의 출력에 반응

@listen() 데코레이터는 Flow 안에서 다른 단계의 출력을 기다리는 리스너 메서드를 표시합니다. 지정한 태스크가 출력을 내면 리스너 메서드가 실행되고, 그 출력을 인자로 받을 수 있어요.

사용 방식은 두 가지입니다.

메서드 이름으로 듣기 — 듣고 싶은 메서드 이름을 문자열로 넘깁니다.
```
@listen("generate_city")
def generate_fun_fact(self, random_city):
    # Implementation
```

메서드 자체로 듣기 — 메서드 객체 자체를 넘깁니다.

@listen(generate_city)
def generate_fun_fact(self, random_city):
    # Implementation

Flow 출력 다루기

Flow의 최종 출력은 마지막으로 완료된 메서드의 출력입니다. kickoff()가 이 최종 출력을 반환해요.

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

class OutputExampleFlow(Flow):
    @start()
    def first_method(self):
        return "Output from first_method"

    @listen(first_method)
    def second_method(self, first_output):
        return f"Second method received: {first_output}"


flow = OutputExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()

print("---- Final Output ----")
print(final_output)

---- Final Output ----
Second method received: Output from first_method

second_method가 마지막으로 완료되므로 그 출력이 Flow의 최종 출력이 되고, kickoff()가 그 값을 반환해요. plot() 메서드는 Flow 구조를 이해하는 데 도움이 되는 HTML 파일을 생성합니다.

상태 접근·갱신

Flow 상태는 서로 다른 메서드가 데이터를 저장·공유하는 통로예요. 실행 후 flow.state로 최종 상태를 확인할 수 있습니다.

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

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

class StateExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        self.state.message = "Hello from first_method"
        self.state.counter += 1

    @listen(first_method)
    def second_method(self):
        self.state.message += " - updated by second_method"
        self.state.counter += 1
        return self.state.message

flow = StateExampleFlow()
flow.plot("my_flow_plot")
final_output = flow.kickoff()
print(f"Final Output: {final_output}")
print("Final State:")
print(flow.state)

Final Output: Hello from first_method - updated by second_method
Final State:
counter=2 message='Hello from first_method - updated by second_method'

Flow 사용량 메트릭

Flow 실행이 끝나면 usage_metrics 속성으로 실행 중 일어난 모든 LLM 호출의 토큰 사용량 합계를 볼 수 있어요. 여기에는 Flow가 오케스트레이션한 모든 Crew의 호출, Agent 도구 안의 호출, Flow 메서드에서 직접 부른 LLM.call(...)까지 포함됩니다. CrewAI Enterprise UI에 표시되는 합계의 SDK 쪽 동일 수치라고 보면 됩니다.

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

class UsageMetricsFlow(Flow):
    @start()
    def run_first_crew(self):
        self.state.first_result = FirstCrew().crew().kickoff()

    @listen(run_first_crew)
    def call_llm_directly(self):
        # Bare LLM call — still counted by flow.usage_metrics
        llm = LLM(model="openai/gpt-4o-mini")
        self.state.summary = llm.call("Summarize the key takeaways.")

    @listen(call_llm_directly)
    def run_second_crew(self):
        self.state.second_result = SecondCrew().crew().kickoff()

flow = UsageMetricsFlow()
flow.kickoff()

print(flow.usage_metrics)
# UsageMetrics(total_tokens=8579, prompt_tokens=6210, completion_tokens=2369,
#              cached_prompt_tokens=0, reasoning_tokens=0,
#              cache_creation_tokens=0, successful_requests=5)

flow.usage_metrics는 flow.kickoff().token_usage와 같지 않아요. 후자는 CrewOutput을 반환한 마지막 @listen 메서드의 CrewOutput.token_usage만 반환해서, 이전 Crew와 순수 LLM.call(...) 호출은 전부 무시합니다. Flow 실행 전체 토큰 합계가 필요하면 flow.usage_metrics를 쓰세요.

UsageMetrics 필드 의미

UsageMetrics 객체는 프로바이더 중립적인 계약을 따릅니다.

Field	Meaning
`total_tokens`	Billed total: `prompt_tokens + completion_tokens`
`prompt_tokens`	Full input/prompt tokens billed for the request
`completion_tokens`	Output/completion tokens billed for the request
`cached_prompt_tokens`	Cache-read subset of prompt tokens (breakdown only)
`cache_creation_tokens`	Cache-write subset of prompt tokens (breakdown only, Anthropic)
`reasoning_tokens`	Reasoning/thinking subset where the provider reports it separately (breakdown only)
`successful_requests`	Number of LLM calls aggregated

cached_prompt_tokens, cache_creation_tokens, reasoning_tokens 같은 breakdown 필드는 total_tokens 위에 더해지는 게 아니라, 이미 prompt_tokens나 completion_tokens에 포함된 부분을 따로 보여 줍니다. 각 UsageMetrics 항목은 한 번의 flow.kickoff() 호출 안에서 일어난 모든 LLM 호출의 합이고, 다음 kickoff() 호출 때 카운터가 리셋됩니다.

Flow 상태 관리

CrewAI Flows는 비구조화(unstructured) 와 구조화(structured) 두 가지 상태 관리 방식을 제공합니다.

비구조화 상태

모든 상태를 Flow 클래스의 state 속성에 저장하는 방식입니다. 엄격한 스키마 없이 필요한 속성을 그때그때 추가·수정할 수 있어서 유연하죠. 비구조화여도 각 상태는 자동으로 고유 UUID를 받습니다.

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

class UnstructuredExampleFlow(Flow):

    @start()
    def first_method(self):
        # The state automatically includes an 'id' field
        print(f"State ID: {self.state['id']}")
        self.state['counter'] = 0
        self.state['message'] = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state['counter'] += 1
        self.state['message'] += " - updated again"

        print(f"State after third_method: {self.state}")


flow = UnstructuredExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

참고: id 필드는 자동으로 생성되어 Flow 실행 내내 보존됩니다. 직접 관리·설정할 필요가 없고, 상태에 새 데이터를 넣어도 유지돼요.

구조화 상태

Pydantic BaseModel 같은 모델로 상태의 정확한 형태를 정의하는 방식입니다. 타입 안전성과 검증, 개발 환경의 자동완성까지 얻을 수 있어요.

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


class ExampleState(BaseModel):
    # Note: 'id' field is automatically added to all states
    counter: int = 0
    message: str = ""


class StructuredExampleFlow(Flow[ExampleState]):

    @start()
    def first_method(self):
        # Access the auto-generated ID if needed
        print(f"State ID: {self.state.id}")
        self.state.message = "Hello from structured flow"

    @listen(first_method)
    def second_method(self):
        self.state.counter += 1
        self.state.message += " - updated"

    @listen(second_method)
    def third_method(self):
        self.state.counter += 1
        self.state.message += " - updated again"

        print(f"State after third_method: {self.state}")


flow = StructuredExampleFlow()
flow.kickoff()

언제 어떤 방식?

비구조화: 상태가 단순하거나 매우 동적일 때, 유연성이 우선일 때, 스키마 정의 부담 없이 빠르게 프로토타이핑할 때.
구조화: 잘 정의되고 일관된 상태 구조가 필요할 때, 타입 안전성과 검증이 중요할 때, IDE 자동완성·타입 체크를 활용하고 싶을 때.

Flow 지속성 (`@persist`)

@persist 데코레이터는 Flow 상태를 자동으로 영속화해, 재시작이나 다른 워크플로 실행 간에도 상태를 유지하게 해 줍니다. 클래스 레벨이면 모든 메서드 상태를, 메서드 레벨이면 특정 메서드만 저장합니다.

@persist  # Using SQLiteFlowPersistence by default
class MyFlow(Flow[MyState]):
    @start()
    def initialize_flow(self):
        # This method will automatically have its state persisted
        self.state.counter = 1
        print("Initialized flow. State ID:", self.state.id)

    @listen(initialize_flow)
    def next_step(self):
        # The state (including self.state.id) is automatically reloaded
        self.state.counter += 1
        print("Flow state is persisted. Counter:", self.state.counter)

저장 상태 Forking

@persist는 kickoff/kickoff_async에서 두 가지 하이드레이션 모드를 지원합니다.

kickoff(inputs={"id": <uuid>}) — resume: 해당 UUID의 최신 스냅샷을 불러와 같은 flow_uuid 아래에서 계속 작성합니다. 히스토리가 이어져요.
kickoff(restore_from_state_id=<uuid>) — fork: 해당 UUID의 최신 스냅샷에서 새 실행의 상태를 채우고, 새 state.id를 부여합니다. 새 실행의 @persist 쓰기는 새 state.id 아래로 들어가고 원래 Flow의 히스토리는 보존됩니다.

restore_from_state_id가 어떤 저장 상태와도 일치하지 않으면 kickoff는 조용히 기본 동작으로 넘어갑니다. restore_from_state_id를 from_checkpoint와 함께 쓰면 ValueError가 나니 둘 중 하나만 하이드레이션 원천으로 쓰세요.

동작 원리

고유 상태 식별 — 각 상태는 자동으로 고유 UUID를 받고, 상태 갱신·메서드 호출을 거쳐도 보존됩니다. 구조화(Pydantic BaseModel)와 비구조화(dict) 상태 모두 지원해요.
기본 SQLite 백엔드 — SQLiteFlowPersistence가 기본 저장 백엔드로, 상태를 로컬 SQLite 데이터베이스에 자동 저장합니다.
오류 처리 — 데이터베이스 연산에 대한 명확한 오류 메시지와 저장·로드 시 자동 상태 검증을 제공합니다.

Flow 제어: 조건 로직 `or_` / `and_`

or_ 함수는 여러 메서드를 듣고, 그중 하나라도 출력을 내면 리스너를 실행합니다.

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

class OrExampleFlow(Flow):

    @start()
    def start_method(self):
        return "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        return "Hello from the second method"

    @listen(or_(start_method, second_method))
    def logger(self, result):
        print(f"Logger: {result}")


flow = OrExampleFlow()
flow.plot("my_flow_plot")
flow.kickoff()

Logger: Hello from the start method
Logger: Hello from the second method

and_ 함수는 지정한 모든 메서드가 출력을 내야만 리스너를 실행합니다.

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

class AndExampleFlow(Flow):

    @start()
    def start_method(self):
        self.state["greeting"] = "Hello from the start method"

    @listen(start_method)
    def second_method(self):
        self.state["joke"] = "What do computers eat? Microchips."

    @listen(and_(start_method, second_method))
    def logger(self):
        print("---- Logger ----")
        print(self.state)

flow = AndExampleFlow()
flow.plot()
flow.kickoff()

---- Logger ----
{'greeting': 'Hello from the start method', 'joke': 'What do computers eat? Microchips.'}

Router — 조건 분기

@router() 데코레이터는 메서드의 출력에 따라 다른 경로로 분기하는 조건 라우팅을 정의합니다.

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

class ExampleState(BaseModel):
    success_flag: bool = False

class RouterFlow(Flow[ExampleState]):

    @start()
    def start_method(self):
        print("Starting the structured flow")
        random_boolean = random.choice([True, False])
        self.state.success_flag = random_boolean

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

    @listen("success")
    def third_method(self):
        print("Third method running")

    @listen("failed")
    def fourth_method(self):
        print("Fourth method running")


flow = RouterFlow()
flow.plot("my_flow_plot")
flow.kickoff()

start_method가 상태에 랜덤 불리언을 넣고, second_method가 그 값으로 "success" 또는 "failed"를 반환하면, @listen("success")/@listen("failed")가 각각 그 라벨에 바인딩되어 실행됩니다.

Human in the Loop (사람 피드백)

@human_feedback 데코레이터는 CrewAI 1.8.0 이상이 필요합니다.

@human_feedback 데코레이터는 Flow 실행을 잠시 멈추고 사람에게 피드백을 받는 휴먼 인 더 루프 워크플로를 가능하게 해 줍니다. 승인 게이트, 품질 리뷰, 사람의 판단이 필요한 결정 지점에 유용해요.

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

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Do you approve this content?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    def generate_content(self):
        return "Content to be reviewed..."

    @listen("approved")
    def on_approval(self, result: HumanFeedbackResult):
        print(f"Approved! Feedback: {result.feedback}")

    @listen("rejected")
    def on_rejection(self, result: HumanFeedbackResult):
        print(f"Rejected. Reason: {result.feedback}")

emit을 지정하면 사람의 자유 형식 피드백을 LLM이 해석해 지정된 결과 중 하나로 축약하고, 그 결과가 해당 @listen 데코레이터를 트리거해요. 피드백 전체는 self.last_human_feedback(가장 최근)나 self.human_feedback_history(전체 목록)로 확인할 수 있습니다. Slack·웹훅 같은 커스텀 프로바이더를 쓰는 비동기/논블로킹 피드백까지 다룬 완전한 가이드는 Human Feedback in Flows를 참고하세요.

Flow에 Agent 추가하기

Agent를 Flow 안에 직접 넣으면, 전체 Crew가 필요 없는 더 가볍고 집중된 태스크 실행이 가능합니다. @listen 메서드가 async라면 await analyst.kickoff_async(query, response_format=MarketAnalysis)처럼 Agent의 비동기 실행을 기다릴 수 있어요. Pydantic 모델로 구조화 출력(response_format)을 지정하면 결과의 pydantic 속성으로 타입 안전하게 받을 수 있습니다.

Flow에 Crew 추가하기

crewai create flow name_of_flow 명령으로 다중 Crew Flow에 필요한 스캐폴딩을 가진 새 CrewAI 프로젝트를 생성할 수 있습니다. 생성된 프로젝트에는 이미 동작하는 poem_crew가 포함되어 있어요.

생성된 구조는 대략 이렇습니다.

Directory/File	Description
`name_of_flow/`	Root directory for the flow.
├── `crews/`	Contains directories for specific crews.
│ └── `poem_crew/`	Directory for the "poem_crew" with its configurations and scripts.
│ ├── `config/`	Configuration files directory for the "poem_crew".
│ │ ├── `agents.yaml`	YAML file defining the agents for "poem_crew".
│ │ └── `tasks.yaml`	YAML file defining the tasks for "poem_crew".
│ ├── `poem_crew.py`	Script for "poem_crew" functionality.
├── `tools/`	Directory for additional tools used in the flow.
│ └── `custom_tool.py`	Custom tool implementation.
├── `main.py`	Main script for running the flow.
├── `README.md`	Project description and instructions.
├── `pyproject.toml`	Configuration file for project dependencies and settings.
└── `.gitignore`	Specifies files and directories to ignore in version control.

main.py에서 Flow 클래스와 @start/@listen 데코레이터로 Crew들을 연결합니다. 흐름은 crewai run (권장) 또는 uv run kickoff으로 실행하고, 의존성은 crewai install 후 .venv/bin/activate로 가상환경을 활성화합니다.

참고: 0.103.0 버전부터 crewai run 명령으로 Flow를 실행할 수 있어요. 프로젝트가 Flow인지(pyproject.toml의 type = "flow" 설정) 자동 감지하고 그에 맞게 실행합니다. 기존 crewai flow kickoff 명령은 deprecated이며, Crew와 Flow 모두 crewai run을 쓰는 걸 권장합니다.

Flow 시각화 (Plot)

flow.plot("my_flow_plot") 메서드로 현재 디렉터리에 my_flow_plot.html 인터랙티브 플롯을 생성할 수 있고, 구조화된 프로젝트 안에서는 crewai flow plot 명령으로 Flow 전체 구성을 시각화할 수 있습니다. 생성된 플롯은 태스크를 노드로, 실행 흐름을 방향 있는 엣지로 표시하며 확대·축소와 노드 호버가 가능합니다.

Flow 실행

Flow를 실행하는 두 가지 방법이 있습니다.

Flow API로 — Flow 클래스 인스턴스를 만들어 kickoff()를 호출합니다.

flow = ExampleFlow()
result = flow.kickoff()

스트리밍 실행 — 출력이 생성되는 대로 실시간으로 받아볼 수 있습니다.

class StreamingFlow(Flow):
    stream = True  # Enable streaming

    @start()
    def research(self):
        # Your flow implementation
        pass

# Iterate over streaming output
flow = StreamingFlow()
streaming = flow.kickoff()
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# Access final result
result = streaming.result

Flow 안의 Memory

모든 Flow는 CrewAI의 통합 Memory 시스템에 자동으로 접근할 수 있어요. 어떤 Flow 메서드 안에서든 세 가지 내장 편의 메서드로 메모리를 저장·회상·추출할 수 있습니다.

Method	Description
`self.remember(content, **kwargs)`	Store content in memory. Accepts optional `scope`, `categories`, `metadata`, `importance`.
`self.recall(query, **kwargs)`	Retrieve relevant memories. Accepts optional `scope`, `categories`, `limit`, `depth`.
`self.extract_memories(content)`	Break raw text into discrete, self-contained memory statements.

Flow 초기화 시 기본 Memory() 인스턴스가 자동 생성되며, 커스텀 메모리를 넘길 수도 있어요.

from crewai.flow.flow import Flow
from crewai import Memory

custom_memory = Memory(
    recency_weight=0.5,
    recency_half_life_days=7,
    embedder={"provider": "ollama", "config": {"model_name": "mxbai-embed-large"}},
)

flow = MyFlow(memory=custom_memory)

메모리는 디스크(LanceDB)에 영속되므로, 이전 실행에서 저장한 사실도 이후 실행에서 회상할 수 있습니다 — 즉 Flow가 시간이 지나며 배우고 지식을 쌓아가는 게 가능해요.

더 알아보기

CrewAI Memory (통합 메모리) — 스코프, 슬라이스, 복합 점수 산정, 임베더 설정
Mastering Flow State Management — 상태 영속·복원 심화
Conversational Flows — 다중 턴 채팅 Flow
Human Feedback in Flows — 플로우 안 휴먼 인 더 루프

CrewAI Flows

CrewAI Flows

본문

Flows가 해결하는 것

시작하기: 랜덤 도시로 재미있는 사실 만들기

@start() — Flow의 시작점

@listen() — 다른 단계의 출력에 반응

Flow 출력 다루기

상태 접근·갱신

Flow 사용량 메트릭

UsageMetrics 필드 의미

Flow 상태 관리

비구조화 상태

구조화 상태

언제 어떤 방식?

Flow 지속성 (@persist)

저장 상태 Forking

동작 원리

Flow 제어: 조건 로직 or_ / and_

Router — 조건 분기

Human in the Loop (사람 피드백)

Flow에 Agent 추가하기

Flow에 Crew 추가하기

Flow 시각화 (Plot)

Flow 실행

Flow 안의 Memory

더 알아보기

`@start()` — Flow의 시작점

`@listen()` — 다른 단계의 출력에 반응

Flow 지속성 (`@persist`)

Flow 제어: 조건 로직 `or_` / `and_`