플로우 상태 정복 (Mastering Flow State)¶

CrewAI Flows에서 상태(state)를 관리하고, 영속화하고, 활용하는 방법을 종합적으로 다루는 가이드입니다. 이 내용을 익히면 견고한 AI 애플리케이션을 만들 수 있어요.

플로우에서 상태(State)의 힘 이해하기¶

상태 관리는 정교한 AI 워크플로우의 토대예요. CrewAI Flows의 상태 시스템을 쓰면 실행 단계 사이에서 컨텍스트를 유지하고, 데이터를 공유하고, 복잡한 애플리케이션 로직을 만들 수 있습니다. 상태 관리를 제대로 익히는 것은 신뢰할 수 있고 유지보수 가능한 강력한 AI 애플리케이션을 만드는 데 필수적이죠.

이 가이드에서는 기본 개념부터 고급 기법까지, 실제 코드 예시와 함께 CrewAI Flows에서 상태를 관리하는 방법을 하나씩 살펴볼게요.

상태 관리가 왜 중요할까요¶

효과적인 상태 관리가 주는 능력은 이렇습니다.

실행 단계 전반에 걸쳐 컨텍스트 유지 — 워크플로우의 여러 단계 사이에서 정보를 매끄럽게 전달할 수 있어요
복잡한 조건부 로직 구현 — 쌓인 데이터를 바탕으로 결정을 내릴 수 있습니다
영속적인 애플리케이션 생성 — 워크플로우 진행 상황을 저장하고 복원할 수 있어요
오류 우아하게 처리 — 더 견고한 애플리케이션을 위한 복구 패턴을 구현할 수 있습니다
애플리케이션 확장 — 적절한 데이터 조직화로 복잡한 워크플로우를 지원할 수 있어요
대화형 애플리케이션 지원 — 컨텍스트에 민감한 AI 상호작용을 위해 대화 기록을 저장하고 접근할 수 있습니다

멀티 턴 채팅(사용자 줄마다 kickoff, ChatState, 의도(intent) 라우팅, 지연 추적, ChatSession)에 대한 내용은 Conversational Flows 문서를 참고하세요.

이제 이런 능력들을 효과적으로 활용하는 방법을 살펴볼게요.

상태 관리 기초¶

플로우 상태 수명주기(The Flow State Lifecycle)¶

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

초기화(Initialization) — 플로우가 생성되면 상태가 초기화됩니다(빈 딕셔너리 또는 Pydantic 모델 인스턴스로요)
수정(Modification) — 플로우 메서드들이 실행되면서 상태를 접근하고 수정해요
전달(Transmission) — 상태가 플로우 메서드들 사이에서 자동으로 전달됩니다
영속화(Persistence, 선택 사항) — 상태를 저장소에 저장했다가 나중에 꺼내올 수 있어요
완료(Completion) — 최종 상태는 실행된 모든 메서드의 누적된 변경 사항을 반영합니다

이 수명주기를 이해하는 것이 효과적인 플로우를 설계하는 핵심이에요.

상태 관리의 두 가지 접근 방식¶

CrewAI는 플로우에서 상태를 관리하는 두 가지 방법을 제공합니다.

비구조화 상태(Unstructured State) — 유연성을 위한 딕셔너리 형태의 객체를 사용해요
구조화 상태(Structured State) — 타입 안전성과 검증을 위한 Pydantic 모델을 사용합니다

각 접근 방식을 자세히 살펴볼게요.

비구조화 상태 관리¶

비구조화 상태는 딕셔너리 형태의 접근 방식을 사용하며, 단순한 애플리케이션에 유연함과 단순함을 제공합니다.

어떻게 작동할까요¶

비구조화 상태에서는 이렇게 동작해요.

self.state를 통해 상태에 접근하는데, 이건 딕셔너리처럼 동작합니다
원하는 시점에 자유롭게 키를 추가, 수정, 제거할 수 있어요
모든 상태는 모든 플로우 메서드에서 자동으로 사용할 수 있습니다

기본 예시¶

비구조화 상태 관리의 간단한 예시를 볼게요.

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 모델로 플로우 상태의 스키마를 정의하며, 타입 안전성, 검증, 더 나은 개발자 경험을 제공합니다.

어떻게 작동할까요¶

구조화 상태에서는 이렇게 동작해요.

상태 구조를 나타내는 Pydantic 모델을 정의합니다
이 모델 타입을 플로우 클래스의 타입 파라미터로 전달해요
self.state를 통해 상태에 접근하는데, 이건 Pydantic 모델 인스턴스처럼 동작합니다
모든 필드는 정의된 타입에 따라 검증됩니다
IDE 자동완성과 타입 검사 지원을 받을 수 있어요

기본 예시¶

구조화 상태 관리를 구현하는 방법을 볼게요.

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}")

구조화 상태의 장점¶

구조화 상태를 쓰면 몇 가지 장점이 있습니다.

타입 안전성(Type Safety) — 개발 시점에 타입 오류를 잡아낼 수 있어요
자기 문서화(Self-Documentation) — 상태 모델이 어떤 데이터가 있는지 명확하게 문서화합니다
검증(Validation) — 데이터 타입과 제약 조건을 자동으로 검증해요
IDE 지원 — 자동완성과 인라인 문서를 받을 수 있습니다
기본값(Default Values) — 누락된 데이터에 대한 폴백을 쉽게 정의할 수 있어요

구조화 상태는 언제 쓸까요¶

구조화 상태는 이런 경우를 권장합니다.

잘 정의된 데이터 스키마를 가진 복잡한 플로우
여러 개발자가 같은 코드를 다루는 팀 프로젝트
데이터 검증이 중요한 애플리케이션
특정 데이터 타입과 제약 조건을 강제해야 하는 플로우

자동 상태 ID(The Automatic State ID)¶

비구조화 상태와 구조화 상태 모두 상태 인스턴스를 추적하고 관리하는 데 도움이 되는 고유 식별자(UUID)를 자동으로 받습니다.

어떻게 작동할까요¶

비구조화 상태에서는 self.state["id"]로 ID에 접근할 수 있어요
구조화 상태에서는 self.state.id로 ID에 접근할 수 있습니다
이 ID는 플로우가 생성될 때 자동으로 생성돼요
플로우의 수명주기 내내 같은 ID가 유지됩니다
ID는 추적, 로깅, 영속화된 상태 검색에 쓸 수 있어요

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

동적 상태 업데이트(Dynamic State Updates)¶

구조화 상태든 비구조화 상태든, 플로우 실행 내내 상태를 동적으로 업데이트할 수 있습니다.

단계 사이에서 데이터 전달하기¶

플로우 메서드가 값을 반환하면, 그 값이 리스닝 메서드의 인자로 전달됩니다.

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})"

이 패턴은 직접 데이터 전달과 상태 업데이트를 결합해서 최대한의 유연성을 얻을 수 있게 해줍니다.

플로우 상태 영속화(Persisting Flow State)¶

CrewAI의 가장 강력한 기능 중 하나는 실행 간에 플로우 상태를 영속화할 수 있다는 점입니다. 이 덕분에 일시 중지, 재개, 실패 후 복구까지 가능한 워크플로우를 만들 수 있어요.

@persist() 데코레이터¶

@persist() 데코레이터는 상태 영속화를 자동화해서, 실행의 핵심 지점에서 플로우 상태를 저장합니다.

클래스 레벨 영속화(Class-Level Persistence)¶

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

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

메서드 레벨 영속화(Method-Level Persistence)¶

더 세밀한 제어가 필요하다면 @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 Persisted State)¶

@persist는 kickoff / kickoff_async에서 두 가지 서로 다른 하이드레이션(hydration) 모드를 지원합니다. resume(inputs["id"])은 같은 계보(lineage)를 이어가는 데 쓰고, fork(restore_from_state_id)는 스냅샷에서 시드된 새 계보를 시작하는 데 사용해요.

	kickoff 후 `state.id`	`@persist` 기록이 저장되는 위치
`inputs["id"]` (resume)	입력된 id	입력된 id (이력 연장)
`restore_from_state_id` (fork)	새 id, 또는 고정되었다면 `inputs["id"]`	새 id (원본 보존)

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의 not-found 동작과 동일해요.
restore_from_state_id를 from_checkpoint와 함께 쓰면 ValueError가 발생합니다. 두 옵션은 서로 다른 상태 시스템(@persist vs. Checkpointing)을 대상으로 하기 때문에 함께 쓸 수 없어요.
restore_from_state_id=None(기본값)은 해당 파라미터 없이 kickoff하는 것과 바이트 단위로 동일합니다.
포크하면서 inputs["id"]를 고정하면 새 실행이 다른 플로우와 영속화 키를 공유하게 됩니다. 보통은 restore_from_state_id만 쓰는 게 좋아요.

고급 상태 패턴(Advanced State Patterns)¶

조건부 시작과 재개 가능한 실행¶

플로우는 HITL(human-in-the-loop)/순환 시나리오를 위해 조건부 @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()는 메서드 이름, 라우터 레이블, 또는 호출 가능한 조건(callable)을 받을 수 있어요.
재개 중에는 리스너들이 이전 체크포인트부터 이어서 실행되고, 순환/라우터 분기들은 재개 플래그를 존중합니다.

상태 기반 조건부 로직¶

상태를 사용해 플로우에서 복잡한 조건부 로직을 구현할 수 있습니다.

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

class PaymentState(BaseModel):
    amount: float = 0.0
    is_approved: bool = False
    retry_count: int = 0

class PaymentFlow(Flow[PaymentState]):
    @start()
    def process_payment(self):
        # Simulate payment processing
        self.state.amount = 100.0
        self.state.is_approved = self.state.amount < 1000
        return "Payment processed"

    @router(process_payment)
    def check_approval(self, previous_result):
        if self.state.is_approved:
            return "approved"
        elif self.state.retry_count < 3:
            return "retry"
        else:
            return "rejected"

    @listen("approved")
    def handle_approval(self):
        return f"Payment of ${self.state.amount} approved!"

    @listen("retry")
    def handle_retry(self):
        self.state.retry_count += 1
        print(f"Retrying payment (attempt {self.state.retry_count})...")
        # Could implement retry logic here
        return "Retry initiated"

    @listen("rejected")
    def handle_rejection(self):
        return f"Payment of ${self.state.amount} rejected after {self.state.retry_count} retries."

복잡한 상태 변환 다루기¶

복잡한 상태 변환에는 전용 메서드를 만들면 됩니다.

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

class UserData(BaseModel):
    name: str
    active: bool = True
    login_count: int = 0

class ComplexState(BaseModel):
    users: Dict[str, UserData] = {}
    active_user_count: int = 0

class TransformationFlow(Flow[ComplexState]):
    @start()
    def initialize(self):
        # Add some users
        self.add_user("alice", "Alice")
        self.add_user("bob", "Bob")
        self.add_user("charlie", "Charlie")
        return "Initialized"

    @listen(initialize)
    def process_users(self, _):
        # Increment login counts
        for user_id in self.state.users:
            self.increment_login(user_id)

        # Deactivate one user
        self.deactivate_user("bob")

        # Update active count
        self.update_active_count()

        return f"Processed {len(self.state.users)} users"

    # Helper methods for state transformations
    def add_user(self, user_id: str, name: str):
        self.state.users[user_id] = UserData(name=name)
        self.update_active_count()

    def increment_login(self, user_id: str):
        if user_id in self.state.users:
            self.state.users[user_id].login_count += 1

    def deactivate_user(self, user_id: str):
        if user_id in self.state.users:
            self.state.users[user_id].active = False
            self.update_active_count()

    def update_active_count(self):
        self.state.active_user_count = sum(
            1 for user in self.state.users.values() if user.active
        )

이렇게 헬퍼 메서드를 만들어 두면 플로우 메서드를 깔끔하게 유지하면서도 복잡한 상태 조작이 가능해집니다.

크루(Crew)와 함께 상태 관리하기¶

CrewAI에서 가장 강력한 패턴 중 하나는 플로우 상태 관리와 크루 실행을 결합하는 것입니다.

크루에 상태 전달하기¶

플로우 상태를 사용해 크루를 파라미터화할 수 있습니다.

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."
        )

        writer = Agent(
            role="Content Writer",
            goal="Transform research into clear, engaging content",
            backstory="You excel at communicating complex ideas clearly and concisely."
        )

        # Create tasks
        research_task = Task(
            description=f"Research {self.state.topic} with {self.state.depth} analysis",
            expected_output="Comprehensive research notes in markdown format",
            agent=researcher
        )

        writing_task = Task(
            description=f"Create a summary on {self.state.topic} based on the research",
            expected_output="Well-written article in markdown format",
            agent=writer,
            context=[research_task]
        )

        # Create and run crew
        research_crew = Crew(
            agents=[researcher, writer],
            tasks=[research_task, writing_task],
            process=Process.sequential,
            verbose=True
        )

        # Run crew and store result in state
        result = research_crew.kickoff()
        self.state.results = result.raw

        return "Research completed"

    @listen(execute_research)
    def summarize_results(self, _):
        # Access the stored results
        result_length = len(self.state.results)
        return f"Research on {self.state.topic} completed with {result_length} characters of results."

크루의 출력을 상태에서 다루기¶

크루가 완료되면, 그 출력을 처리해서 플로우 상태에 저장할 수 있습니다.

@listen(execute_crew)
def process_crew_results(self, _):
    # Parse the raw results (assuming JSON output)
    import json
    try:
        results_dict = json.loads(self.state.raw_results)
        self.state.processed_results = {
            "title": results_dict.get("title", ""),
            "main_points": results_dict.get("main_points", []),
            "conclusion": results_dict.get("conclusion", "")
        }
        return "Results processed successfully"
    except json.JSONDecodeError:
        self.state.error = "Failed to parse crew results as JSON"
        return "Error processing results"

상태 관리 모범 사례(Best Practices for State Management)¶

1. 상태는 집중적으로 유지하기¶

필요한 것만 담도록 상태를 설계하세요.

# Too broad
class BloatedState(BaseModel):
    user_data: Dict = {}
    system_settings: Dict = {}
    temporary_calculations: List = []
    debug_info: Dict = {}
    # ...many more fields

# Better: Focused state
class FocusedState(BaseModel):
    user_id: str
    preferences: Dict[str, str]
    completion_status: Dict[str, bool]

2. 복잡한 플로우에는 구조화 상태 사용하기¶

플로우가 복잡해질수록 구조화 상태의 가치가 커집니다.

# Simple flow can use unstructured state
class SimpleGreetingFlow(Flow):
    @start()
    def greet(self):
        self.state["name"] = "World"
        return f"Hello, {self.state['name']}!"

# Complex flow benefits from structured state
class UserRegistrationState(BaseModel):
    username: str
    email: str
    verification_status: bool = False
    registration_date: datetime = Field(default_factory=datetime.now)
    last_login: Optional[datetime] = None

class RegistrationFlow(Flow[UserRegistrationState]):
    # Methods with strongly-typed state access

3. 상태 전이 문서화하기¶

복잡한 플로우라면 실행 내내 상태가 어떻게 변하는지 문서화해 두세요.

@start()
def initialize_order(self):
    """
    Initialize order state with empty values.

    State before: {}
    State after: {order_id: str, items: [], status: 'new'}
    """
    self.state.order_id = str(uuid.uuid4())
    self.state.items = []
    self.state.status = "new"
    return "Order initialized"

4. 상태 오류는 우아하게 처리하기¶

상태 접근에 대한 오류 처리를 구현해 두세요.

@listen(previous_step)
def process_data(self, _):
    try:
        # Try to access a value that might not exist
        user_preference = self.state.preferences.get("theme", "default")
    except (AttributeError, KeyError):
        # Handle the error gracefully
        self.state.errors = self.state.get("errors", [])
        self.state.errors.append("Failed to access preferences")
        user_preference = "default"

    return f"Used preference: {user_preference}"

5. 진행 상황 추적에 상태 사용하기¶

오래 실행되는 플로우에서 진행 상황을 추적하는 데 상태를 활용하세요.

class ProgressTrackingFlow(Flow):
    @start()
    def initialize(self):
        self.state["total_steps"] = 3
        self.state["current_step"] = 0
        self.state["progress"] = 0.0
        self.update_progress()
        return "Initialized"

    def update_progress(self):
        """Helper method to calculate and update progress"""
        if self.state.get("total_steps", 0) > 0:
            self.state["progress"] = (self.state.get("current_step", 0) /
                                    self.state["total_steps"]) * 100
            print(f"Progress: {self.state['progress']:.1f}%")

    @listen(initialize)
    def step_one(self, _):
        # Do work...
        self.state["current_step"] = 1
        self.update_progress()
        return "Step 1 complete"

    # Additional steps...

6. 가능하면 불변 연산 사용하기¶

특히 구조화 상태에서는 명확성을 위해 불변 연산을 선호하는 게 좋습니다.

# Instead of modifying lists in place:
self.state.items.append(new_item)  # Mutable operation

# Consider creating new state:
from pydantic import BaseModel
from typing import List

class ItemState(BaseModel):
    items: List[str] = []

class ImmutableFlow(Flow[ItemState]):
    @start()
    def add_item(self):
        # Create new list with the added item
        self.state.items = [*self.state.items, "new item"]
        return "Item added"

플로우 상태 디버깅(Debugging Flow State)¶

상태 변경 로깅하기¶

개발할 때는 상태 변경을 추적하는 로깅을 추가해 두세요.

import logging
logging.basicConfig(level=logging.INFO)

class LoggingFlow(Flow):
    def log_state(self, step_name):
        logging.info(f"State after {step_name}: {self.state}")

    @start()
    def initialize(self):
        self.state["counter"] = 0
        self.log_state("initialize")
        return "Initialized"

    @listen(initialize)
    def increment(self, _):
        self.state["counter"] += 1
        self.log_state("increment")
        return f"Incremented to {self.state['counter']}"

상태 시각화하기¶

디버깅을 위해 상태를 시각화하는 메서드를 추가할 수 있습니다.

def visualize_state(self):
    """Create a simple visualization of the current state"""
    import json
    from rich.console import Console
    from rich.panel import Panel

    console = Console()

    if hasattr(self.state, "model_dump"):
        # Pydantic v2
        state_dict = self.state.model_dump()
    elif hasattr(self.state, "dict"):
        # Pydantic v1
        state_dict = self.state.dict()
    else:
        # Unstructured state
        state_dict = dict(self.state)

    # Remove id for cleaner output
    if "id" in state_dict:
        state_dict.pop("id")

    state_json = json.dumps(state_dict, indent=2, default=str)
    console.print(Panel(state_json, title="Current Flow State"))

마무리¶

CrewAI Flows에서 상태 관리를 마스터하면, 컨텍스트를 유지하고 복잡한 결정을 내리며 일관된 결과를 내는 정교하고 견고한 AI 애플리케이션을 만들 수 있습니다. 비구조화 상태를 선택하든 구조화 상태를 선택하든, 올바른 상태 관리 방식을 구현하면 유지보수 가능하고 확장 가능하며 실제 문제를 해결하는 데 효과적인 플로우가 만들어져요.

더 복잡한 플로우를 개발하다 보면 기억하세요. 좋은 상태 관리는 유연성과 구조 사이의 올바른 균형을 찾는 일이고, 그 덕분에 코드가 강력하면서도 이해하기 쉬워집니다.

이제 CrewAI Flows에서 상태 관리의 개념과 실무를 익혔습니다. 이 지식을 바탕으로 컨텍스트를 효과적으로 유지하고 단계 간 데이터를 공유하며 정교한 애플리케이션 로직을 구축하는 견고한 AI 워크플로우를 만들 수 있을 거예요.

다음 단계(Next Steps)¶

플로우에서 구조화 상태와 비구조화 상태를 모두 실험해 보세요
오래 실행되는 워크플로우를 위해 상태 영속화를 구현해 보세요
첫 번째 크루 만들기를 살펴보면서 크루와 플로우가 함께 어떻게 작동하는지 확인해 보세요
Flow 레퍼런스 문서를 확인해 더 고급 기능을 알아보세요