> ## Documentation Index > Fetch the complete documentation index at: https://crewai.burt.pe.kr/llms.txt > Use this file to discover all available pages before exploring further. # Flows > CrewAI Flows를 사용하여 AI 워크플로우를 생성하고 관리하는 방법을 알아보세요. ## 소개 CrewAI Flows는 AI 워크플로우 생성과 관리를 간소화하도록 설계된 강력한 기능이다. Flows는 개발자가 코딩 작업과 Crews를 효율적으로 결합하고 조정할 수 있게 해주며, 정교한 AI 자동화를 구축하기 위한 견고한 프레임워크를 제공한다. Flows를 사용하면 구조화된 이벤트 기반 워크플로우를 만들 수 있다. 이를 통해 여러 작업을 원활하게 연결하고 상태를 관리하며 AI 애플리케이션의 실행 흐름을 제어할 수 있다. Flows를 활용하면 CrewAI의 기능을 최대한 활용하는 다단계 프로세스를 쉽게 설계하고 구현할 수 있다. 1. **워크플로우 생성 간소화**: 여러 Crews와 작업을 쉽게 연결해 복잡한 AI 워크플로우를 생성한다. 2. **상태 관리**: 워크플로우 내에서 서로 다른 작업 간의 상태를 쉽게 관리하고 공유할 수 있다. 3. **이벤트 기반 아키텍처**: 이벤트 기반 모델을 기반으로 동적이고 반응형 워크플로우를 지원한다. 4. **유연한 제어 흐름**: 조건부 논리, 루프, 분기 등을 워크플로우 내에서 구현할 수 있다. ## 시작하기 OpenAI를 사용해 무작위 도시를 생성하고, 그 도시에 대한 재미있는 사실을 생성하는 간단한 Flow를 만들어보자. ```python theme={null} from crewai.flow.flow import Flow, listen, start from dotenv import load_dotenv from litellm import completion class ExampleFlow(Flow): model = "gpt-4o-mini" @start() def generate_city(self): print("Flow 시작") # 각 Flow 상태는 자동으로 고유 ID를 가짐 print(f"Flow 상태 ID: {self.state['id']}") response = completion( model=self.model, messages=[ { "role": "user", "content": "세계에서 무작위 도시 이름 하나를 반환하세요.", }, ], ) random_city = response["choices"][0]["message"]["content"] # 도시를 상태에 저장 self.state["city"] = random_city print(f"무작위 도시: {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"{random_city}에 대한 재미있는 사실을 알려주세요.", }, ], ) fun_fact = response["choices"][0]["message"]["content"] # 재미있는 사실을 상태에 저장 self.state["fun_fact"] = fun_fact return fun_fact flow = ExampleFlow() result = flow.kickoff() print(f"생성된 재미있는 사실: {result}") ``` 위 예제에서는 OpenAI를 사용해 무작위 도시를 생성하고, 그 도시에 대한 재미있는 사실을 생성하는 간단한 Flow를 만들었다. 이 Flow는 `generate_city`와 `generate_fun_fact` 두 가지 작업으로 구성된다. `generate_city` 작업은 Flow의 시작점이고, `generate_fun_fact` 작업은 `generate_city` 작업의 출력을 기다린다. 각 Flow 인스턴스는 자동으로 상태에 고유 식별자(UUID)를 받는다. 이 식별자는 Flow 실행을 추적하고 관리하는 데 도움이 된다. 상태는 Flow 실행 동안 지속되는 추가 데이터(생성된 도시와 재미있는 사실 등)도 저장할 수 있다. Flow를 실행하면 다음과 같은 일이 발생한다: 1. Flow 상태에 고유 ID를 생성 2. 무작위 도시를 생성하고 상태에 저장 3. 그 도시에 대한 재미있는 사실을 생성하고 상태에 저장 4. 결과를 콘솔에 출력 상태의 고유 ID와 저장된 데이터는 Flow 실행을 추적하고 작업 간 컨텍스트를 유지하는 데 유용하다. **참고:** `.env` 파일에 `OPENAI_API_KEY`를 설정했는지 확인한다. 이 키는 OpenAI API 요청을 인증하는 데 필요하다. ### @start() `@start()` 데코레이터는 메서드를 Flow의 시작점으로 표시한다. Flow가 시작되면 `@start()` 데코레이터가 붙은 모든 메서드가 병렬로 실행된다. 하나의 Flow에서 여러 개의 시작 메서드를 가질 수 있으며, Flow가 시작될 때 이들 모두가 실행된다. ### @listen() `@listen()` 데코레이터는 플로우 내 다른 태스크의 출력을 수신하는 메서드를 표시한다. `@listen()`으로 데코레이팅된 메서드는 지정된 태스크가 출력을 내보낼 때 실행된다. 이 메서드는 수신한 태스크의 출력에 접근할 수 있다. #### 사용법 `@listen()` 데코레이터는 여러 가지 방식으로 사용할 수 있다: 1. **메서드 이름으로 리스닝**: 리스닝할 메서드 이름을 문자열로 전달한다. 해당 메서드가 완료되면 리스너 메서드가 트리거된다. ```python theme={null} @listen("generate_city") def generate_fun_fact(self, random_city): # 구현 ``` 2. **메서드 직접 리스닝**: 메서드 자체를 전달한다. 해당 메서드가 완료되면 리스너 메서드가 트리거된다. ```python theme={null} @listen(generate_city) def generate_fun_fact(self, random_city): # 구현 ``` ### Flow 출력 처리 Flow의 출력에 접근하고 처리하는 것은 AI 워크플로를 더 큰 애플리케이션이나 시스템에 통합하는 데 필수적이다. CrewAI Flows는 최종 출력을 가져오고, 중간 결과에 접근하며, Flow의 전반적인 상태를 관리할 수 있는 직관적인 메커니즘을 제공한다. #### 최종 결과 가져오기 Flow를 실행할 때, 마지막으로 완료되는 메서드가 최종 결과를 결정한다. `kickoff()` 메서드는 이 최종 메서드의 결과를 반환한다. 최종 결과에 접근하는 방법은 다음과 같다: ```python Code theme={null} from crewai.flow.flow import Flow, listen, start class OutputExampleFlow(Flow): @start() def first_method(self): return "첫 번째 메서드의 결과" @listen(first_method) def second_method(self, first_output): return f"두 번째 메서드가 받은 결과: {first_output}" flow = OutputExampleFlow() final_output = flow.kickoff() print("---- 최종 결과 ----") print(final_output) ``` ```text Output theme={null} ---- 최종 결과 ---- 두 번째 메서드가 받은 결과: 첫 번째 메서드의 결과 ``` 이 예제에서 `second_method`가 마지막으로 완료되는 메서드이므로, 이 메서드의 결과가 Flow의 최종 결과가 된다. `kickoff()` 메서드는 최종 결과를 반환하며, 이 결과는 콘솔에 출력된다. #### 상태 접근 및 업데이트 최종 출력을 가져오는 것 외에도, Flow 내부의 상태에 접근하고 업데이트할 수 있다. 상태는 Flow 내의 다양한 메서드 간에 데이터를 저장하고 공유하는 데 사용된다. Flow가 실행된 후, 상태에 접근하여 실행 중 추가되거나 업데이트된 정보를 가져올 수 있다. 다음은 상태를 업데이트하고 접근하는 방법을 보여주는 예제다: ```python Code theme={null} 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() final_output = flow.kickoff() print(f"Final Output: {final_output}") print("Final State:") print(flow.state) ``` ```text Output theme={null} Final Output: Hello from first_method - updated by second_method Final State: counter=2 message='Hello from first_method - updated by second_method' ``` 이 예제에서 `first_method`와 `second_method`가 상태를 업데이트한다. Flow가 실행된 후, 최종 상태에 접근하여 이 메서드들이 수행한 업데이트를 확인할 수 있다. 최종 메서드의 출력을 반환하고 상태에 접근할 수 있도록 함으로써, CrewAI Flow는 AI 워크플로우의 결과를 더 큰 애플리케이션이나 시스템에 통합하기 쉽게 한다. 또한 Flow 실행 중 상태를 유지하고 접근할 수 있다. ## 플로우 상태 관리 효과적인 상태 관리는 신뢰할 수 있고 유지보수가 용이한 AI 워크플로우를 구축하는 데 필수적이다. CrewAI Flows는 비정형 및 정형 상태 관리를 위한 강력한 메커니즘을 제공한다. 이를 통해 개발자는 애플리케이션의 요구사항에 가장 적합한 접근 방식을 선택할 수 있다. ### 비정형 상태 관리 비정형 상태 관리에서는 모든 상태가 `Flow` 클래스의 `state` 속성에 저장된다. 이 방식은 개발자가 엄격한 스키마를 정의하지 않고도 상태 속성을 동적으로 추가하거나 수정할 수 있는 유연성을 제공한다. 비정형 상태를 사용하더라도 CrewAI Flows는 각 상태 인스턴스에 대해 고유 식별자(UUID)를 자동으로 생성하고 유지한다. ```python theme={null} from crewai.flow.flow import Flow, listen, start class UnstructuredExampleFlow(Flow): @start() def first_method(self): # 상태에 자동으로 'id' 필드가 포함된다 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.kickoff() ``` **참고:** `id` 필드는 자동으로 생성되며, 플로우 실행 전반에 걸쳐 유지된다. 이 필드를 수동으로 관리하거나 설정할 필요가 없으며, 새로운 데이터로 상태를 업데이트할 때도 유지된다. **핵심 포인트:** * **유연성:** `self.state`에 미리 정의된 제약 없이 동적으로 속성을 추가할 수 있다. * **단순성:** 상태 구조가 최소화되거나 크게 변동되는 간단한 워크플로우에 적합하다. ### 구조화된 상태 관리 구조화된 상태 관리는 미리 정의된 스키마를 활용해 워크플로우 전반에 걸쳐 일관성과 타입 안전성을 보장한다. Pydantic의 `BaseModel`과 같은 모델을 사용하면 상태의 정확한 구조를 정의할 수 있어, 개발 환경에서 더 나은 검증과 자동 완성 기능을 활용할 수 있다. CrewAI Flows의 각 상태는 자동으로 고유 식별자(UUID)를 받아 상태 인스턴스를 추적하고 관리한다. 이 ID는 Flow 시스템에 의해 자동으로 생성되고 관리된다. ```python theme={null} from crewai.flow.flow import Flow, listen, start from pydantic import BaseModel class ExampleState(BaseModel): # 참고: 모든 상태에 'id' 필드가 자동으로 추가됨 counter: int = 0 message: str = "" class StructuredExampleFlow(Flow[ExampleState]): @start() def first_method(self): # 필요 시 자동 생성된 ID에 접근 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() ``` **핵심 포인트:** * **정의된 스키마:** `ExampleState`는 상태 구조를 명확히 정의해 코드의 가독성과 유지보수성을 높인다. * **타입 안전성:** Pydantic을 활용하면 상태 속성이 지정된 타입을 준수하도록 보장해 런타임 오류를 줄인다. * **자동 완성:** IDE는 정의된 상태 모델을 기반으로 더 나은 자동 완성과 오류 검사를 제공할 수 있다. ### 비구조적 상태 관리와 구조적 상태 관리 중 선택하기 * **비구조적 상태 관리를 사용하는 경우:** * 워크플로우의 상태가 단순하거나 매우 동적일 때 * 엄격한 상태 정의보다 유연성이 더 중요할 때 * 스키마 정의에 드는 오버헤드 없이 빠르게 프로토타이핑이 필요할 때 * **구조적 상태 관리를 사용하는 경우:** * 워크플로우가 잘 정의되고 일관된 상태 구조를 요구할 때 * 애플리케이션의 신뢰성을 위해 타입 안전성과 검증이 중요할 때 * 개발자 경험을 향상시키기 위해 자동 완성 및 타입 검사와 같은 IDE 기능을 활용하고 싶을 때 CrewAI Flows는 비구조적과 구조적 상태 관리 옵션을 모두 제공함으로써 개발자들이 유연하면서도 견고한 AI 워크플로우를 구축할 수 있도록 지원한다. 이를 통해 다양한 애플리케이션 요구사항에 맞춰 작업할 수 있다. ## 상태 유지 기능 @persist 데코레이터는 CrewAI 플로우에서 자동 상태 유지 기능을 활성화한다. 이를 통해 플로우를 재시작하거나 다른 워크플로 실행 간에도 상태를 유지할 수 있다. 이 데코레이터는 클래스 수준이나 메서드 수준 모두에서 적용 가능하며, 상태 유지 방식을 유연하게 관리할 수 있다. ### 클래스 수준의 상태 유지 @persist 데코레이터를 클래스 수준에 적용하면 모든 플로우 메서드의 상태가 자동으로 유지된다: ```python theme={null} @persist # 기본적으로 SQLiteFlowPersistence 사용 class MyFlow(Flow[MyState]): @start() def initialize_flow(self): # 이 메서드는 자동으로 상태가 유지됨 self.state.counter = 1 print("플로우 초기화 완료. 상태 ID:", self.state.id) @listen(initialize_flow) def next_step(self): # 상태(self.state.id 포함)가 자동으로 다시 로드됨 self.state.counter += 1 print("플로우 상태가 유지됨. 카운터:", self.state.counter) ``` ### 메서드 단위 지속성 더 세밀한 제어를 위해 특정 메서드에 @persist를 적용할 수 있다: ```python theme={null} class AnotherFlow(Flow[dict]): @persist # 이 메서드의 상태만 지속 @start() def begin(self): if "runs" not in self.state: self.state["runs"] = 0 self.state["runs"] += 1 print("메서드 단위로 지속된 실행 횟수:", self.state["runs"]) ``` ### 동작 원리 1. **고유 상태 식별** * 각 플로우 상태는 자동으로 고유한 UUID를 받는다. * 이 ID는 상태 업데이트와 메서드 호출 시에도 유지된다. * 구조화된(Pydantic BaseModel) 상태와 비구조화된(딕셔너리) 상태를 모두 지원한다. 2. **기본 SQLite 백엔드** * SQLiteFlowPersistence가 기본 스토리지 백엔드로 사용된다. * 상태는 로컬 SQLite 데이터베이스에 자동으로 저장된다. * 데이터베이스 작업이 실패할 경우 명확한 오류 메시지를 제공한다. 3. **오류 처리** * 데이터베이스 작업에 대한 포괄적인 오류 메시지를 제공한다. * 저장 및 로드 시 자동으로 상태를 검증한다. * 지속성 작업 중 문제가 발생하면 명확한 피드백을 제공한다. ### 주요 고려사항 * **상태 타입**: 구조화된 상태(Pydantic BaseModel)와 비구조화된 상태(딕셔너리) 모두 지원한다. * **자동 ID**: `id` 필드가 없을 경우 자동으로 추가된다. * **상태 복구**: 실패하거나 재시작된 플로우는 이전 상태를 자동으로 다시 불러올 수 있다. * **커스텀 구현**: 특수한 저장소 요구 사항이 있을 경우 직접 FlowPersistence 구현체를 제공할 수 있다. ### 기술적 장점 1. **저수준 접근을 통한 정밀한 제어** * 고급 사용 사례를 위한 지속성 작업에 직접 접근 가능 * 메서드 수준의 지속성 데코레이터를 통한 세밀한 제어 * 내장된 상태 검사 및 디버깅 기능 * 상태 변화와 지속성 작업에 대한 완전한 가시성 2. **향상된 신뢰성** * 시스템 장애 또는 재시작 후 자동 상태 복구 * 데이터 무결성을 위한 트랜잭션 기반 상태 업데이트 * 명확한 오류 메시지를 포함한 포괄적인 오류 처리 * 상태 저장 및 로드 작업 중 강력한 검증 3. **확장 가능한 아키텍처** * FlowPersistence 인터페이스를 통한 커스텀 가능한 지속성 백엔드 * SQLite 외의 특수 저장 솔루션 지원 * 구조화된(Pydantic) 및 비구조화된(dict) 상태 모두와 호환 * 기존 CrewAI 플로우 패턴과의 원활한 통합 지속성 시스템의 아키텍처는 기술적 정밀성과 커스터마이징 옵션을 강조한다. 이를 통해 개발자는 내장된 신뢰성 기능의 이점을 누리면서도 상태 관리에 대한 완전한 제어를 유지할 수 있다. ## 플로우 제어 ### 조건부 로직: `or` Flows의 `or_` 함수는 여러 메서드를 동시에 감시하고, 지정된 메서드 중 하나라도 출력을 내면 리스너 메서드를 실행한다. ```python Code theme={null} 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.kickoff() ``` ```text Output theme={null} Logger: Hello from the start method Logger: Hello from the second method ``` 이 플로우를 실행하면 `logger` 메서드는 `start_method` 또는 `second_method` 중 하나의 출력에 의해 트리거된다. `or_` 함수는 여러 메서드를 감시하고, 지정된 메서드 중 하나라도 출력을 내면 리스너 메서드를 실행하는 데 사용된다. ### 조건부 로직: `and` Flows에서 `and_` 함수를 사용하면 여러 메서드를 동시에 감시하고, 지정된 모든 메서드가 출력을 내보낼 때만 리스너 메서드를 트리거할 수 있다. ```python Code theme={null} 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.kickoff() ``` ```text Output theme={null} ---- Logger ---- {'greeting': 'Hello from the start method', 'joke': 'What do computers eat? Microchips.'} ``` 이 Flow를 실행하면 `logger` 메서드는 `start_method`와 `second_method` 모두가 출력을 내보낼 때만 트리거된다. `and_` 함수는 여러 메서드를 감시하고, 지정된 모든 메서드가 출력을 내보낼 때만 리스너 메서드를 실행하는 데 사용된다. ### 라우터 `@router()` 데코레이터를 사용하면 메서드의 출력값에 따라 조건부 라우팅 로직을 정의할 수 있다. 이를 통해 메서드의 결과에 따라 다른 경로를 지정하고, 실행 흐름을 동적으로 제어할 수 있다. ```python Code theme={null} 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.kickoff() ``` ```text Output theme={null} Starting the structured flow Third method running Fourth method running ``` 위 예제에서 `start_method`는 임의의 불리언 값을 생성해 상태에 설정한다. `second_method`는 `@router()` 데코레이터를 사용해 불리언 값에 따라 조건부 라우팅 로직을 정의한다. 불리언 값이 `True`면 `"success"`를 반환하고, `False`면 `"failed"`를 반환한다. `third_method`와 `fourth_method`는 `second_method`의 출력을 기다렸다가 반환된 값에 따라 실행된다. 이 플로우를 실행하면 `start_method`에서 생성된 임의의 불리언 값에 따라 출력이 달라진다. ## 플로우에 크루 추가하기 CrewAI에서 여러 크루를 포함한 플로우를 만드는 작업은 간단하다. 다음 커맨드를 실행하면 여러 크루를 포함한 플로우를 만들기 위해 필요한 모든 기본 구조가 포함된 새로운 CrewAI 프로젝트를 생성할 수 있다: ```bash theme={null} crewai create flow name_of_flow ``` 이 커맨드는 필요한 폴더 구조와 함께 새로운 CrewAI 프로젝트를 생성한다. 생성된 프로젝트에는 이미 작동 중인 `poem_crew`라는 사전 제작된 크루가 포함되어 있다. 이 크루를 템플릿으로 사용해 복사, 붙여넣기, 수정 작업을 통해 다른 크루를 만들 수 있다. ### 폴더 구조 `crewai create flow name_of_flow` 명령어를 실행한 후, 다음과 같은 폴더 구조를 확인할 수 있다: | 디렉토리/파일 | 설명 | | :--------------------- | :--------------------------------------------- | | `name_of_flow/` | 플로우의 루트 디렉토리. | | ├── `crews/` | 특정 크루를 위한 디렉토리. | | │ └── `poem_crew/` | "poem\_crew"를 위한 디렉토리. 여기에는 설정 파일과 스크립트가 포함된다. | | │ ├── `config/` | "poem\_crew"의 설정 파일 디렉토리. | | │ │ ├── `agents.yaml` | "poem\_crew"의 에이전트를 정의하는 YAML 파일. | | │ │ └── `tasks.yaml` | "poem\_crew"의 태스크를 정의하는 YAML 파일. | | │ ├── `poem_crew.py` | "poem\_crew"의 기능을 구현한 스크립트. | | ├── `tools/` | 플로우에서 사용되는 추가 도구를 위한 디렉토리. | | │ └── `custom_tool.py` | 커스텀 도구 구현 파일. | | ├── `main.py` | 플로우를 실행하는 메인 스크립트. | | ├── `README.md` | 프로젝트 설명과 사용법을 담은 파일. | | ├── `pyproject.toml` | 프로젝트 의존성과 설정을 위한 구성 파일. | | └── `.gitignore` | 버전 관리에서 무시할 파일과 디렉토리를 지정하는 파일. | ### 크루 구성하기 `crews` 폴더 안에서 여러 크루를 정의할 수 있다. 각 크루는 설정 파일과 크루 정의 파일을 포함하는 고유의 폴더를 가진다. 예를 들어, `poem_crew` 폴더에는 다음 파일들이 있다: * `config/agents.yaml`: 크루에 속한 에이전트를 정의한다. * `config/tasks.yaml`: 크루에 속한 태스크를 정의한다. * `poem_crew.py`: 에이전트, 태스크, 그리고 크루 자체를 포함하는 크루 정의 파일이다. `poem_crew`를 복사하고 붙여넣어 다른 크루를 만들 수 있다. 필요한 부분을 수정하면 된다. ### `main.py`에서 크루 연결하기 `main.py` 파일은 여러분의 플로우를 생성하고 크루를 연결하는 곳이다. `Flow` 클래스와 `@start`, `@listen` 데코레이터를 사용해 실행 흐름을 정의할 수 있다. 다음은 `main.py` 파일에서 `poem_crew`를 연결하는 예제이다: ```python theme={null} #!/usr/bin/env python from random import randint from pydantic import BaseModel from crewai.flow.flow import Flow, listen, start from .crews.poem_crew.poem_crew import PoemCrew class PoemState(BaseModel): sentence_count: int = 1 poem: str = "" class PoemFlow(Flow[PoemState]): @start() def generate_sentence_count(self): print("문장 수 생성 중") self.state.sentence_count = randint(1, 5) @listen(generate_sentence_count) def generate_poem(self): print("시 생성 중") result = PoemCrew().crew().kickoff(inputs={"sentence_count": self.state.sentence_count}) print("시 생성 완료", result.raw) self.state.poem = result.raw @listen(generate_poem) def save_poem(self): print("시 저장 중") with open("poem.txt", "w") as f: f.write(self.state.poem) def kickoff(): poem_flow = PoemFlow() poem_flow.kickoff() def plot(): poem_flow = PoemFlow() poem_flow.plot() if __name__ == "__main__": kickoff() ``` 이 예제에서 `PoemFlow` 클래스는 문장 수를 생성하고, `PoemCrew`를 사용해 시를 생성한 후, 이를 파일로 저장하는 플로우를 정의한다. 플로우는 `kickoff()` 메서드를 호출해 시작한다. ### 플로우 실행하기 (선택 사항) 플로우를 실행하기 전에, 다음 명령어를 실행해 필요한 의존성을 설치한다: ```bash theme={null} crewai install ``` 의존성 설치가 완료되면, 가상 환경을 활성화해야 한다. 다음 명령어를 실행한다: ```bash theme={null} source .venv/bin/activate ``` 가상 환경을 활성화한 후, 다음 명령어 중 하나를 실행해 플로우를 실행한다: ```bash theme={null} crewai flow kickoff ``` 또는 ```bash theme={null} uv run kickoff ``` 플로우가 실행되면, 콘솔에서 결과를 확인할 수 있다. ## 플롯 흐름 시각화 AI 워크플로우를 시각화하면 흐름의 구조와 실행 경로를 더 잘 이해할 수 있다. CrewAI는 강력한 시각화 도구를 제공해 여러분의 흐름을 인터랙티브한 플롯으로 생성할 수 있게 한다. 이를 통해 AI 워크플로우를 더 쉽게 이해하고 최적화할 수 있다. ### 플롯이란 무엇인가? CrewAI에서 플롯은 AI 워크플로우를 시각적으로 표현한 그래픽이다. 각각의 작업, 그들 간의 연결, 그리고 데이터의 흐름을 보여준다. 이 시각화는 작업의 순서를 이해하고, 병목 현상을 식별하며, 워크플로우 로직이 기대한 대로 작동하는지 확인하는 데 도움을 준다. ### 플롯 생성 방법 CrewAI는 여러분의 작업 흐름을 시각화하기 위한 두 가지 편리한 방법을 제공한다: #### Option 1: `plot()` 메서드 사용하기 플로우 인스턴스를 직접 다룰 때는 `plot()` 메서드를 호출해 플롯을 생성할 수 있다. 이 메서드는 플로우의 인터랙티브 플롯을 포함한 HTML 파일을 만든다. ```python theme={null} # 플로우 인스턴스가 있다고 가정 flow.plot("my_flow_plot") ``` 이 코드는 현재 디렉토리에 `my_flow_plot.html` 파일을 생성한다. 웹 브라우저에서 이 파일을 열어 인터랙티브 플롯을 확인할 수 있다. #### 옵션 2: 커맨드라인 사용 구조화된 CrewAI 프로젝트 내에서 작업한다면, 커맨드라인을 통해 플로우 다이어그램을 생성할 수 있다. 이 방법은 전체 플로우 설정을 시각적으로 확인하고 싶은 대규모 프로젝트에 특히 유용하다. ```bash theme={null} crewai flow plot ``` 이 커맨드를 실행하면 `plot()` 메서드와 유사한 플로우 다이어그램이 포함된 HTML 파일이 생성된다. 파일은 프로젝트 디렉토리에 저장되며, 웹 브라우저에서 열어 플로우를 탐색할 수 있다. ### 플롯 이해하기 생성된 플롯은 여러분의 흐름에 있는 작업을 나타내는 노드와 실행 흐름을 나타내는 방향성 간선으로 구성된다. 이 플롯은 상호작용이 가능하며, 확대/축소가 가능하고 노드 위에 마우스를 올리면 추가 세부 정보를 확인할 수 있다. 흐름을 시각화하면 워크플로우의 구조를 더 명확히 이해할 수 있다. 이를 통해 디버깅, 최적화, 그리고 AI 프로세스를 다른 사람들에게 설명하기가 더 쉬워진다. ### 결론 CrewAI의 플로우 시각화 기능은 복잡한 AI 워크플로우를 설계하고 관리하는 데 큰 도움이 된다. `plot()` 메서드를 사용하든 커맨드라인을 통해 플롯을 생성하든, 워크플로우를 시각적으로 표현하면 개발과 프레젠테이션 모두에 유용하다. ## 다음 단계 추가적인 플로우 예제를 탐색하고 싶다면, 예제 저장소에서 다양한 추천 사례를 확인할 수 있다. 여기서는 네 가지 구체적인 플로우 예제를 소개하며, 각각 고유한 사용 사례를 보여준다. 이를 통해 현재 문제 유형에 맞는 예제를 찾는 데 도움을 받을 수 있다. 1. **이메일 자동 응답 플로우**: 이 예제는 백그라운드 작업이 지속적으로 실행되어 이메일 응답을 자동화하는 무한 루프를 보여준다. 수동 개입 없이 반복적으로 수행해야 하는 작업에 적합한 사례이다. [예제 보기](https://github.com/crewAIInc/crewAI-examples/tree/main/email_auto_responder_flow) 2. **리드 스코어 플로우**: 이 플로우는 인간의 피드백을 포함하고 라우터를 사용해 다양한 조건부 분기를 처리하는 방법을 보여준다. 동적 의사 결정과 인간의 감독을 워크플로우에 통합하는 방법을 배울 수 있는 훌륭한 예제이다. [예제 보기](https://github.com/crewAIInc/crewAI-examples/tree/main/lead-score-flow) 3. **책 작성 플로우**: 이 예제는 여러 크루를 연결하는 데 탁월하다. 한 크루가 책의 전체 개요를 작성하면, 다른 크루가 그 개요를 바탕으로 각 장을 생성한다. 결국 모든 것이 연결되어 완전한 책을 만든다. 복잡한 다단계 프로세스와 작업 간 조정이 필요한 경우에 적합한 플로우이다. [예제 보기](https://github.com/crewAIInc/crewAI-examples/tree/main/write_a_book_with_flows) 4. **회의 어시스턴트 플로우**: 이 플로우는 하나의 이벤트를 브로드캐스트하여 여러 후속 작업을 트리거하는 방법을 보여준다. 예를 들어, 회의가 끝난 후 Trello 보드를 업데이트하고, Slack 메시지를 보내고, 결과를 저장하는 등의 작업을 수행할 수 있다. 단일 이벤트에서 여러 결과를 처리하는 방법을 배울 수 있는 훌륭한 예제로, 포괄적인 작업 관리 및 알림 시스템에 적합하다. [예제 보기](https://github.com/crewAIInc/crewAI-examples/tree/main/meeting_assistant_flow) 이 예제들을 탐색하면 CrewAI 플로우를 다양한 사용 사례에 활용하는 방법을 깊이 있게 이해할 수 있다. 반복 작업 자동화부터 동적 의사 결정과 인간 피드백이 필요한 복잡한 다단계 프로세스 관리까지 다양한 시나리오에 적용할 수 있다. 또한, CrewAI에서 플로우를 사용하는 방법에 대한 YouTube 비디오도 확인해 보자!