Mastering Flow State Management

Mastering Flow State Management (플로우 상태 관리 마스터하기)

복잡한 AI 워크플로에서 상태 관리가 잘 안 되면 단계 사이의 맥락이 끊기고, 실패한 실행을 다시 이어가기도 어려워져요. CrewAI Flows의 상태 시스템은 단계 사이에 데이터를 공유하고, 실행 재개가 가능한 영속 앱을 만들고, 대화형 앱의 히스토리를 유지하게 해 줍니다. 이 가이드에서는 상태의 기본부터 구조화·비구조화 상태, 상태 영속과 @persist, 그리고 Crew와 결합하는 고급 패턴까지 실전 코드와 함께 정리합니다.

출처: 공식문서

본문

상태 관리가 중요한 이유

효과적인 상태 관리를 통해 다음을 할 수 있습니다.

실행 단계 간 맥락 유지 — 워크플로의 서로 다른 단계 사이에 정보를 매끄럽게 전달
복잡한 조건 로직 구현 — 쌓인 데이터를 바탕으로 결정
영속 애플리케이션 생성 — 워크플로 진행 상황을 저장·복원
오류 우아하게 처리 — 더 견고한 앱을 위한 복구 패턴
애플리케이션 확장 — 적절한 데이터 구성으로 복잡한 워크플로 지원
대화형 애플리케이션 지원 — 대화 히스토리를 저장·접근해 맥락 있는 AI 상호작용

다중 턴 채팅(kickoff per user line, ChatState, 의도 라우팅, 지연 트레이싱, ChatSession)에 관해서는 Conversational Flows를 참고하세요.

Flow 상태 수명주기

CrewAI Flows에서 상태는 예측 가능한 수명주기를 따릅니다.

초기화 — Flow가 생성될 때 상태가 초기화됨 (빈 dict 또는 Pydantic 모델 인스턴스)
수정 — Flow 메서드가 실행되며 상태에 접근·수정
전달 — 상태가 Flow 메서드 사이에 자동 전달
영속 (선택) — 상태를 저장소에 저장했다가 나중에 검색
완료 — 최종 상태가 실행된 모든 메서드의 누적 변경을 반영

두 가지 상태 관리 방식

CrewAI는 Flow 상태를 관리하는 두 가지 방식을 제공합니다.

비구조화 상태 (Unstructured) — 딕셔너리형 객체로 유연성 확보
구조화 상태 (Structured) — Pydantic 모델로 타입 안전성과 검증

비구조화 상태 관리

비구조화 상태는 딕셔너리형 접근 방식을 사용합니다. self.state가 딕셔너리처럼 동작해 키를 언제든 자유롭게 추가·수정·삭제할 수 있고, 모든 상태는 모든 Flow 메서드에서 자동으로 접근 가능해요.

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

class UnstructuredStateFlow(Flow):
    @start()
    def initialize_data(self):
        print("Initializing flow data")
        # Add key-value pairs to state
        self.state["user_name"] = "Alex"
        self.state["preferences"] = {
            "theme": "dark",
            "language": "English"
        }
        self.state["items"] = []

        # The flow state automatically gets a unique ID
        print(f"Flow ID: {self.state['id']}")

        return "Initialized"

    @listen(initialize_data)
    def process_data(self, previous_result):
        print(f"Previous step returned: {previous_result}")

        # Access and modify state
        user = self.state["user_name"]
        print(f"Processing data for {user}")

        # Add items to a list in state
        self.state["items"].append("item1")
        self.state["items"].append("item2")

        # Add a new key-value pair
        self.state["processed"] = True

        return "Processed"

    @listen(process_data)
    def generate_summary(self, previous_result):
        # Access multiple state values
        user = self.state["user_name"]
        theme = self.state["preferences"]["theme"]
        items = self.state["items"]
        processed = self.state.get("processed", False)

        summary = f"User {user} has {len(items)} items with {theme} theme. "
        summary += "Data is processed." if processed else "Data is not processed."

        return summary

# Run the flow
flow = UnstructuredStateFlow()
result = flow.kickoff()
print(f"Final result: {result}")
print(f"Final state: {flow.state}")

비구조화 상태는 빠른 프로토타이핑, 동적으로 진화하는 상태, 구조를 미리 모르는 경우, 단순한 상태 요구 사항에 적합합니다. 다만 타입 체크와 스키마 검증이 없어 복잡한 앱에서는 오류가 발생할 수 있어요.

구조화 상태 관리

구조화 상태는 Pydantic 모델로 Flow 상태의 스키마를 정의합니다. 타입 안전성, 검증, 더 나은 개발 경험을 제공해요.

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel, Field
from typing import List, Dict, Optional

# Define your state model
class UserPreferences(BaseModel):
    theme: str = "light"
    language: str = "English"

class AppState(BaseModel):
    user_name: str = ""
    preferences: UserPreferences = UserPreferences()
    items: List[str] = []
    processed: bool = False
    completion_percentage: float = 0.0

# Create a flow with typed state
class StructuredStateFlow(Flow[AppState]):
    @start()
    def initialize_data(self):
        print("Initializing flow data")
        # Set state values (type-checked)
        self.state.user_name = "Taylor"
        self.state.preferences.theme = "dark"

        # The ID field is automatically available
        print(f"Flow ID: {self.state.id}")

        return "Initialized"

    @listen(initialize_data)
    def process_data(self, previous_result):
        print(f"Processing data for {self.state.user_name}")

        # Modify state (with type checking)
        self.state.items.append("item1")
        self.state.items.append("item2")
        self.state.processed = True
        self.state.completion_percentage = 50.0

        return "Processed"

    @listen(process_data)
    def generate_summary(self, previous_result):
        # Access state (with autocompletion)
        summary = f"User {self.state.user_name} has {len(self.state.items)} items "
        summary += f"with {self.state.preferences.theme} theme. "
        summary += "Data is processed." if self.state.processed else "Data is not processed."
        summary += f" Completion: {self.state.completion_percentage}%"

        return summary

# Run the flow
flow = StructuredStateFlow()
result = flow.kickoff()
print(f"Final result: {result}")
print(f"Final state: {flow.state}")

구조화 상태의 장점은 ① 개발 시점에 타입 오류를 잡는 타입 안전성, ② 어떤 데이터가 있는지 명확히 문서화되는 자기 문서화, ③ 데이터 타입·제약의 자동 검증, ④ IDE의 자동완성·인라인 문서, ⑤ 누락 데이터에 대한 기본값 정의입니다. 스키마가 잘 정의된 복잡한 Flow, 다수 개발자가 작업하는 팀 프로젝트, 데이터 검증이 중요한 앱에 권장됩니다.

자동 상태 ID

비구조화·구조화 상태 모두 자동으로 고유 식별자(UUID)를 받습니다.

비구조화 상태: self.state["id"]
구조화 상태: self.state.id
Flow 생성 시 자동 생성되고, Flow 수명주기 전반에 걸쳐 동일하게 유지됩니다.
추적·로깅·영속 상태 검색에 사용 가능

이 UUID는 영속화를 구현하거나 여러 Flow 실행을 추적할 때 특히 유용합니다.

단계 간 데이터 전달

Flow 메서드는 값을 반환할 수 있고, 이 값이 리스닝 메서드의 인자로 전달됩니다.

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

class DataPassingFlow(Flow):
    @start()
    def generate_data(self):
        # This return value will be passed to listening methods
        return "Generated data"

    @listen(generate_data)
    def process_data(self, data_from_previous_step):
        print(f"Received: {data_from_previous_step}")
        # You can modify the data and pass it along
        processed_data = f"{data_from_previous_step} - processed"
        # Also update state
        self.state["last_processed"] = processed_data
        return processed_data

    @listen(process_data)
    def finalize_data(self, processed_data):
        print(f"Received processed data: {processed_data}")
        # Access both the passed data and state
        last_processed = self.state.get("last_processed", "")
        return f"Final: {processed_data} (from state: {last_processed})"

이 패턴으로 직접 데이터 전달과 상태 갱신을 결합해 최대한의 유연성을 얻을 수 있습니다.

Flow 상태 영속화 (`@persist`)

CrewAI의 강력한 기능 중 하나는 Flow 상태를 실행 간에 영속화하는 것입니다. 잠시 멈췄다 재개하거나, 실패 후 복구할 수 있는 워크플로를 가능하게 해 줍니다.

클래스 레벨 영속 — @persist()를 Flow 클래스 전체에 적용하면 매 메서드 실행 후 상태를 저장합니다.

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

class CounterState(BaseModel):
    value: int = 0

@persist()  # Apply to the entire flow class
class PersistentCounterFlow(Flow[CounterState]):
    @start()
    def increment(self):
        self.state.value += 1
        print(f"Incremented to {self.state.value}")
        return self.state.value

    @listen(increment)
    def double(self, value):
        self.state.value = value * 2
        print(f"Doubled to {self.state.value}")
        return self.state.value

# First run
flow1 = PersistentCounterFlow()
result1 = flow1.kickoff()
print(f"First run result: {result1}")

# Second run - pass the ID to load the persisted state
flow2 = PersistentCounterFlow()
result2 = flow2.kickoff(inputs={"id": flow1.state.id})
print(f"Second run result: {result2}")  # Will be higher due to persisted state

메서드 레벨 영속 — 더 세밀하게 제어하려면 @persist()를 특정 메서드에만 적용합니다.

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

class SelectivePersistFlow(Flow):
    @start()
    def first_step(self):
        self.state["count"] = 1
        return "First step"

    @persist()  # Only persist after this method
    @listen(first_step)
    def important_step(self, prev_result):
        self.state["count"] += 1
        self.state["important_data"] = "This will be persisted"
        return "Important step completed"

    @listen(important_step)
    def final_step(self, prev_result):
        self.state["count"] += 1
        return f"Complete with count {self.state['count']}"

저장 상태 Forking — @persist는 kickoff/kickoff_async에서 두 가지 하이드레이션 모드를 지원합니다. resume은 inputs["id"]로 같은 계보를 이어가고, fork는 restore_from_state_id로 스냅샷에서 새 계보를 시작합니다.

	`state.id` after kickoff	`@persist` writes land under
`inputs["id"]` (resume)	supplied id	supplied id (extends history)
`restore_from_state_id` (fork)	fresh id, or `inputs["id"]` if pinned	new id (source preserved)

from crewai.flow.flow import Flow, start
from crewai.flow.persistence import persist
from pydantic import BaseModel

class CounterState(BaseModel):
    id: str = ""
    counter: int = 0

@persist
class CounterFlow(Flow[CounterState]):
    @start()
    def step(self):
        self.state.counter += 1

# Run 1: fresh state, counter 0 -> 1
flow_1 = CounterFlow()
flow_1.kickoff()

# Fork: hydrate from flow_1's latest snapshot, but write under a NEW state.id
flow_2 = CounterFlow()
flow_2.kickoff(restore_from_state_id=flow_1.state.id)
# flow_2 starts with counter=1 (hydrated), then step() bumps it to 2.
# flow_1's flow_uuid history is unchanged.

동작 참고:

restore_from_state_id가 영속 저장에서 발견되지 않으면 kickoff가 조용히 기본 동작으로 폴백합니다(기존 inputs["id"] resume 미발견 동작과 동일). 예외가 발생하지 않아요.
restore_from_state_id를 from_checkpoint와 함께 쓰면 ValueError가 납니다 — 서로 다른 상태 시스템(@persist vs Checkpointing)을 대상으로 하므로 함께 쓸 수 없어요.
restore_from_state_id=None(기본)은 파라미터 없는 kickoff와 바이트 단위로 동일합니다.
forking 중에 inputs["id"]를 고정하면 새 실행이 다른 Flow와 영속 키를 공유하게 됩니다 — 보통 restore_from_state_id만 쓰는 게 좋아요.

고급 상태 패턴

조건부 시작과 재개 가능한 실행 — Flow는 HITL/순환 시나리오를 위한 조건부 @start()와 재개 가능한 실행을 지원합니다.

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

class ResumableFlow(Flow):
    @start()  # unconditional start
    def init(self):
        ...

    # Conditional start: run after "init" or external trigger name
    @start("init")
    def maybe_begin(self):
        ...

    @listen(and_(init, maybe_begin))
    def proceed(self):
        ...

조건부 @start()는 메서드 이름, 라우터 라벨, 또는 호출 가능한 조건을 받습니다.
재개 중에 리스너는 이전 체크포인트부터 계속되고, 사이클/라우터 브랜치는 재개 플래그를 존중합니다.

상태 기반 조건 로직 — 상태를 사용해 복잡한 조건 로직을 구현할 수 있습니다. 예를 들어 결제 승인 플로우에서 @router로 승인/재시도/거절 분기를 만들 수 있어요. self.state.is_approved, self.state.retry_count 같은 상태 값을 읽어 라우팅 결정에 활용합니다.

복잡한 상태 변환 — 복잡한 상태 변환에는 전용 헬퍼 메서드를 만드는 패턴이 좋습니다. add_user, increment_login, deactivate_user 같은 헬퍼를 두면 Flow 메서드를 깔끔하게 유지하면서 복잡한 상태 조작을 가능하게 해요.

상태 관리와 Crew 결합

가장 강력한 패턴 중 하나는 Flow 상태 관리를 크루 실행과 결합하는 것입니다. Flow 상태로 크루를 파라미터화할 수 있습니다 — 예를 들어 ResearchState의 topic·depth 필드를 크루의 에이전트 goal과 태스크 description에 반영하는 식입니다.

from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

class ResearchState(BaseModel):
    topic: str = ""
    depth: str = "medium"
    results: str = ""

class ResearchFlow(Flow[ResearchState]):
    @start()
    def get_parameters(self):
        # In a real app, this might come from user input
        self.state.topic = "Artificial Intelligence Ethics"
        self.state.depth = "deep"
        return "Parameters set"

    @listen(get_parameters)
    def execute_research(self, _):
        # Create agents
        researcher = Agent(
            role="Research Specialist",
            goal=f"Research {self.state.topic} in {self.state.depth} detail",
            backstory="You are an expert researcher with a talent for finding accurate information."
        )
        # ... create writer agent, tasks, and crew, then crew.kickoff() ...

        return "Research complete"

이렇게 하면 Flow의 상태가 크루 구성의 단일 진실 원천(source of truth)이 되어, 사용자 입력부터 각 에이전트의 목표와 태스크까지 한 흐름으로 이어집니다.

더 알아보기

CrewAI Flows — Flow 개념·데코레이터·상태 기초
Conversational Flows — 다중 턴 채팅 상태
Human Feedback in Flows — 사람 피드백을 상태·라우팅에 연결
Mastering Flow State 공식 문서

Mastering Flow State Management