Post

CrewAI Flows의 상태 관리 마스터하기

CrewAI Flows에서 상태 관리를 활용하는 기본 개념과 비구조화 상태 관리 예제를 소개합니다.

Posted
By mminzy22
views 28 min read
CrewAI Flows의 상태 관리 마스터하기

Flow에서 상태의 힘 이해하기

상태 관리는 복잡한 AI 워크플로우의 중추입니다. CrewAI Flows에서는 상태 시스템을 통해 문맥을 유지하고, 단계 간 데이터를 공유하며, 복잡한 애플리케이션 로직을 구성할 수 있습니다. 상태 관리를 마스터하는 것은 신뢰할 수 있고 유지보수가 쉬우며 강력한 AI 애플리케이션을 만드는 데 필수적입니다.

이 가이드는 CrewAI Flows에서 상태를 관리하는 기본 개념부터 고급 기술까지 실제 코드 예제와 함께 설명합니다.

왜 상태 관리가 중요한가?

효과적인 상태 관리는 다음을 가능하게 합니다:

  1. 단계 간 문맥 유지 - 다양한 단계 사이에서 데이터를 원활하게 전달
  2. 복잡한 조건 로직 구성 - 누적된 데이터를 기반으로 의사 결정
  3. 지속 가능한 애플리케이션 구축 - 워크플로우 진행 상황 저장 및 복원
  4. 에러 복구 패턴 구현 - 오류가 발생해도 회복 가능한 플로우 설계
  5. 애플리케이션 확장성 향상 - 상태 데이터를 체계적으로 관리하여 복잡한 흐름 지원
  6. 대화형 애플리케이션 구성 - 대화 기록을 저장하여 문맥 기반 응답 제공

이제 이러한 기능들을 어떻게 활용할 수 있을지 살펴보겠습니다.

상태 관리의 기본

Flow 상태의 생명주기

CrewAI Flows에서 상태는 다음과 같은 생명주기를 가집니다:

  1. 초기화 - 플로우가 생성되면 상태가 빈 딕셔너리 혹은 Pydantic 모델 인스턴스로 초기화됨
  2. 수정 - 각 단계에서 상태에 접근하고 값을 변경
  3. 전파 - 상태는 자동으로 다음 단계로 전달됨
  4. 영속화 (선택적) - 필요시 상태를 저장소에 저장하고 나중에 불러올 수 있음
  5. 완료 - 마지막 단계까지 수행된 상태가 최종 상태가 됨

이러한 흐름을 이해하는 것이 안정적인 플로우 설계의 핵심입니다.

상태 관리 방식 두 가지

CrewAI는 상태를 다음 두 가지 방식으로 관리할 수 있도록 지원합니다:

  1. 비구조화 상태(Unstructured State) - 딕셔너리와 유사한 구조로 유연하게 사용
  2. 구조화 상태(Structured State) - Pydantic 모델을 활용한 타입 안전성 및 검증 지원

이번 글에서는 먼저 비구조화 상태 방식에 대해 자세히 살펴보겠습니다.

비구조화 상태 관리

비구조화 상태는 딕셔너리 형태로 간단하고 유연한 상태 저장 방식을 제공합니다. 특히 프로토타이핑이나 단순한 로직에 유용합니다.

어떻게 동작하나?

  • 상태는 self.state를 통해 접근하며, 이는 딕셔너리처럼 작동합니다
  • 어떤 키든 자유롭게 추가, 수정, 삭제할 수 있습니다
  • 모든 단계에서 동일한 상태를 공유합니다

기본 예제

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

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

class UnstructuredStateFlow(Flow):
    @start()
    def initialize_data(self):
        print("Initializing flow data")  # 플로우 데이터 초기화 시작
        # 상태에 키-값 쌍 추가
        self.state["user_name"] = "Alex"  # 사용자 이름 설정
        self.state["preferences"] = {
            "theme": "dark",  # 테마 설정
            "language": "English"  # 언어 설정
        }
        self.state["items"] = []  # 아이템 목록 초기화

        # 플로우 상태에는 고유 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}")  # 이전 단계 결과 출력

        # 상태 접근 및 수정
        user = self.state["user_name"]
        print(f"Processing data for {user}")  # 사용자 이름 기반 메시지 출력

        # 상태 내 리스트에 아이템 추가
        self.state["items"].append("item1")
        self.state["items"].append("item2")

        # 새로운 키-값 쌍 추가
        self.state["processed"] = True
        return "Processed"

    @listen(process_data)
    def generate_summary(self, previous_result):
        # 여러 상태 값 접근
        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

# 플로우 실행
flow = UnstructuredStateFlow()
result = flow.kickoff()
print(f"Final result: {result}")  # 최종 결과 출력
print(f"Final state: {flow.state}")  # 최종 상태 출력

비구조화 상태를 사용할 때

비구조화 상태는 다음과 같은 경우에 적합합니다:

  • 빠른 프로토타입 개발 시
  • 구조가 미리 정해지지 않은 상태일 때
  • 단순한 로직의 흐름을 다룰 때

단, 타입 검사나 자동 완성 기능이 부족하므로, 복잡한 데이터 구조를 다루는 데는 부적합할 수 있습니다.

구조화 상태 관리

구조화 상태는 Pydantic 모델을 사용해 상태의 스키마를 정의함으로써, 타입 안정성, 검증, 그리고 더 나은 개발 경험을 제공합니다.

어떻게 동작하나?

  • Pydantic 모델을 정의하여 상태 구조를 명시합니다
  • 이 모델을 Flow 클래스에 타입 인자로 전달합니다
  • self.state를 통해 Pydantic 모델 인스턴스처럼 상태에 접근합니다
  • 모든 필드는 정의된 타입에 따라 자동 검증됩니다
  • IDE에서 자동완성 및 타입 검사 기능을 활용할 수 있습니다

기본 예제

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

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

# 상태 모델 정의
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

# 타입이 명시된 상태를 사용하는 Flow 생성
class StructuredStateFlow(Flow[AppState]):
    @start()
    def initialize_data(self):
        print("Initializing flow data")  # 플로우 데이터 초기화
        # 상태 값 설정 (타입 검증 포함)
        self.state.user_name = "Taylor"
        self.state.preferences.theme = "dark"

        # ID 필드는 자동으로 제공됨
        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}")  # 상태에서 이름 접근

        # 상태 수정 (타입 검증 포함)
        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):
        # 상태 접근 (자동완성 지원)
        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

# 플로우 실행
flow = StructuredStateFlow()
result = flow.kickoff()
print(f"Final result: {result}")
print(f"Final state: {flow.state}")

구조화 상태의 장점

구조화 상태를 사용하면 다음과 같은 이점이 있습니다:

  1. 타입 안정성(Type Safety) - 개발 시점에서 타입 오류를 사전에 방지
  2. 자체 문서화(Self-Documentation) - 상태 모델이 어떤 데이터가 사용되는지 명확히 설명
  3. 자동 검증(Validation) - 데이터 타입 및 제약 조건 자동 검증
  4. IDE 지원 - 자동완성 및 인라인 문서 지원
  5. 기본값 설정(Default Values) - 누락된 값에 대해 기본값 제공 용이

구조화 상태를 사용할 때

구조화 상태는 다음과 같은 경우에 권장됩니다:

  • 명확한 데이터 스키마가 필요한 복잡한 플로우
  • 여러 개발자가 동시에 작업하는 팀 프로젝트
  • 데이터 검증이 중요한 애플리케이션
  • 특정 타입과 제약 조건을 강제해야 하는 경우

자동 생성되는 상태 ID

비구조화 상태든 구조화 상태든, CrewAI Flow의 상태 객체는 자동으로 고유 식별자(UUID)를 부여받습니다. 이는 상태를 추적하고 관리하는 데 매우 유용합니다.

어떻게 동작하나?

  • 비구조화 상태에서는 self.state["id"]로 접근 가능
  • 구조화 상태에서는 self.state.id로 접근 가능
  • 이 ID는 플로우 생성 시 자동으로 부여됩니다
  • 플로우 실행 전체 과정에서 동일한 ID가 유지됩니다
  • 상태 저장/복원 시 또는 로그 추적, 분석 등에 활용할 수 있습니다

이 UUID는 상태 영속성 기능을 구현하거나 여러 플로우 실행을 구분할 때 특히 유용합니다.

동적인 상태 업데이트

구조화 상태이든 비구조화 상태이든 관계없이, Flow 실행 도중 상태를 동적으로 업데이트할 수 있습니다.

단계 간 데이터 전달하기

Flow의 메서드는 값을 반환할 수 있고, 이 값은 이후 단계의 메서드에서 인자로 전달받을 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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

class DataPassingFlow(Flow):
    @start()
    def generate_data(self):
        # 이 반환값은 다음 단계 메서드에 인자로 전달됨
        return "Generated data"

    @listen(generate_data)
    def process_data(self, data_from_previous_step):
        print(f"Received: {data_from_previous_step}")  # 전달받은 데이터 출력
        # 데이터를 수정하여 전달할 수 있음
        processed_data = f"{data_from_previous_step} - processed"
        # 상태도 함께 업데이트 가능
        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}")  # 처리된 데이터 출력
        # 전달받은 데이터와 상태를 동시에 활용 가능
        last_processed = self.state.get("last_processed", "")
        return f"Final: {processed_data} (from state: {last_processed})"

이 패턴을 사용하면 직접적인 데이터 전달과 상태 업데이트를 병행하여 유연한 데이터 흐름을 구현할 수 있습니다.

상태 영속화 (Persistence)

CrewAI의 강력한 기능 중 하나는 상태를 플로우 실행 간에 유지(저장)할 수 있다는 점입니다. 이를 통해 중단된 워크플로우를 재개하거나 오류 이후 복구할 수 있습니다.

@persist 데코레이터

@persist 데코레이터는 상태를 자동으로 저장하여, 지정된 시점마다 플로우 상태를 지속시킵니다.

클래스 수준에서의 상태 저장

클래스에 @persist를 붙이면, 모든 메서드 실행 후 상태가 자동 저장됩니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

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

class CounterState(BaseModel):
    value: int = 0

@persist  # 클래스 전체에 적용
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

# 첫 번째 실행
flow1 = PersistentCounterFlow()
result1 = flow1.kickoff()
print(f"First run result: {result1}")

# 두 번째 실행 - 상태는 자동으로 불러와짐
flow2 = PersistentCounterFlow()
result2 = flow2.kickoff()
print(f"Second run result: {result2}")  # 저장된 상태로 인해 더 높은 값 반환됨

메서드 수준에서의 상태 저장

보다 세밀한 제어를 원할 경우, @persist를 특정 메서드에만 적용할 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

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

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

    @persist  # 이 메서드 실행 후에만 상태 저장
    @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']}"

이처럼 @persist는 전체 플로우 또는 특정 지점에서의 상태 보존을 유연하게 설정할 수 있는 기능입니다.

고급 상태 패턴

상태 기반 조건 분기 로직

상태를 활용해 플로우 내 복잡한 조건 분기를 구현할 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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):
        # 결제 처리 시뮬레이션
        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})...")
        # 재시도 로직을 이곳에 구현할 수 있음
        return "Retry initiated"

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

복잡한 상태 변환 처리

복잡한 상태 변형이 필요한 경우, 보조 메서드를 따로 정의해 메인 플로우를 깔끔하게 유지할 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

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):
        # 사용자 추가
        self.add_user("alice", "Alice")
        self.add_user("bob", "Bob")
        self.add_user("charlie", "Charlie")
        return "Initialized"

    @listen(initialize)
    def process_users(self, _):
        # 로그인 수 증가
        for user_id in self.state.users:
            self.increment_login(user_id)

        # 한 명 비활성화
        self.deactivate_user("bob")

        # 활성 사용자 수 갱신
        self.update_active_count()

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

    # 상태 변환을 위한 헬퍼 메서드들
    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에서 가장 강력한 패턴 중 하나는 Flow의 상태 관리와 Crew 실행을 결합하는 것입니다.

상태를 Crew에 전달하기

Flow의 상태를 활용해 Crew에 파라미터를 전달할 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

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):
        # 실제 애플리케이션에서는 사용자 입력을 받을 수도 있음
        self.state.topic = "Artificial Intelligence Ethics"
        self.state.depth = "deep"
        return "Parameters set"

    @listen(get_parameters)
    def execute_research(self, _):
        # 에이전트 생성
        researcher = Agent(
            role="Research Specialist",
            goal=f"Research {self.state.topic} in {self.state.depth} detail",
            backstory="정확한 정보를 찾아내는 능력을 가진 전문 연구자입니다."
        )

        writer = Agent(
            role="Content Writer",
            goal="연구 내용을 명확하고 흥미롭게 정리합니다",
            backstory="복잡한 개념을 쉽게 전달하는 데 능숙한 작가입니다."
        )

        # 태스크 생성
        research_task = Task(
            description=f"{self.state.topic} 주제를 {self.state.depth} 수준으로 분석",
            expected_output="마크다운 형식의 연구 노트",
            agent=researcher
        )

        writing_task = Task(
            description=f"연구 내용을 바탕으로 {self.state.topic}에 대한 요약 작성",
            expected_output="마크다운 형식의 잘 작성된 기사",
            agent=writer,
            context=[research_task]
        )

        # 크루 구성 및 실행
        research_crew = Crew(
            agents=[researcher, writer],
            tasks=[research_task, writing_task],
            process=Process.sequential,
            verbose=True
        )

        # 크루 실행 및 결과를 상태에 저장
        result = research_crew.kickoff()
        self.state.results = result.raw

        return "Research completed"

    @listen(execute_research)
    def summarize_results(self, _):
        # 저장된 결과 길이 출력
        result_length = len(self.state.results)
        return f"{self.state.topic} 주제에 대한 연구가 {result_length}자의 결과로 완료되었습니다."

Crew 출력값을 상태에 저장하기

Crew 실행 후 그 결과를 파싱하여 상태에 저장할 수도 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@listen(execute_crew)
def process_crew_results(self, _):
    # 결과가 JSON 형식이라고 가정하고 파싱
    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 = "Crew 결과를 JSON으로 파싱하는 데 실패했습니다"
        return "Error processing results"

상태 관리를 위한 모범 사례

1. 상태는 꼭 필요한 정보만 담기

상태를 설계할 때는 반드시 필요한 정보만 포함하도록 설계합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13

# 과도하게 많은 상태 예시
class BloatedState(BaseModel):
    user_data: Dict = {}
    system_settings: Dict = {}
    temporary_calculations: List = []
    debug_info: Dict = {}
    # 기타 필드들

# 더 나은 예시: 집중된 상태 구조
class FocusedState(BaseModel):
    user_id: str
    preferences: Dict[str, str]
    completion_status: Dict[str, bool]

2. 복잡한 Flow에는 구조화 상태를 쓰기

플로우가 복잡해질수록 구조화된 상태 모델이 더욱 유용해집니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 단순한 플로우는 비구조화 상태로도 충분함
class SimpleGreetingFlow(Flow):
    @start()
    def greet(self):
        self.state["name"] = "World"
        return f"Hello, {self.state['name']}!"

# 복잡한 플로우에는 구조화 상태 추천
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]):
    # 구조화 상태를 활용한 강력한 타입 기반 접근
    pass

3. 상태 변화 문서화하기

복잡한 플로우일수록, 상태가 어떻게 변화하는지를 주석으로 명시해 두는 것이 좋습니다.

1
2
3
4
5
6
7
8
9
10
11
12

@start()
def initialize_order(self):
    """
    주문 상태를 빈 값으로 초기화합니다.

    초기 상태: {}
    이후 상태: {order_id: str, items: [], status: 'new'}
    """
    self.state.order_id = str(uuid.uuid4())
    self.state.items = []
    self.state.status = "new"
    return "Order initialized"

4. 상태 접근 오류를 우아하게 처리하기

상태 값에 접근할 때 오류가 발생할 수 있으므로, 예외 처리를 통해 문제를 예방합니다.

1
2
3
4
5
6
7
8
9
10
11
12

@listen(previous_step)
def process_data(self, _):
    try:
        # 존재하지 않을 수도 있는 값 접근 시도
        user_preference = self.state.preferences.get("theme", "default")
    except (AttributeError, KeyError):
        # 오류 발생 시 graceful하게 처리
        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. 진행 상황 추적에 상태 활용하기

장기 실행되는 플로우에서는 상태를 통해 진행 상황을 추적할 수 있습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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):
        """진행률 계산 및 출력"""
        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, _):
        # 실제 작업 수행
        self.state["current_step"] = 1
        self.update_progress()
        return "Step 1 complete"

6. 가능하면 불변 방식(Immutable)을 사용하기

특히 구조화된 상태에서는 가변(mutable) 방식보다 불변 방식을 선호하는 것이 좋습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 리스트를 직접 수정하는 대신:
self.state.items.append(new_item)  # 가변 방식

# 새로운 리스트로 상태를 갱신하는 방식:
from pydantic import BaseModel
from typing import List

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

class ImmutableFlow(Flow[ItemState]):
    @start()
    def add_item(self):
        # 새 항목이 포함된 새 리스트 생성
        self.state.items = [*self.state.items, "new item"]
        return "Item added"

상태 디버깅

상태 변경 로그 출력하기

개발 단계에서는 상태 변경 내역을 로그로 출력해보는 것이 좋습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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']}"

상태 시각화

디버깅 시 현재 상태를 시각적으로 출력하는 함수도 유용합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

def visualize_state(self):
    """현재 상태를 보기 좋게 출력"""
    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:
        # 비구조화 상태
        state_dict = dict(self.state)

    # ID는 출력에서 생략
    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
TIL CrewAI Crew Flow
This post is licensed under CC BY 4.0 by the author.