​
소개

CrewAI Flows는 AI 워크플로우 생성과 관리를 간소화하도록 설계된 강력한 기능이다. Flows는 개발자가 코딩 작업과 Crews를 효율적으로 결합하고 조정할 수 있게 해주며, 정교한 AI 자동화를 구축하기 위한 견고한 프레임워크를 제공한다.

Flows를 사용하면 구조화된 이벤트 기반 워크플로우를 만들 수 있다. 이를 통해 여러 작업을 원활하게 연결하고 상태를 관리하며 AI 애플리케이션의 실행 흐름을 제어할 수 있다. Flows를 활용하면 CrewAI의 기능을 최대한 활용하는 다단계 프로세스를 쉽게 설계하고 구현할 수 있다.

  1. 워크플로우 생성 간소화: 여러 Crews와 작업을 쉽게 연결해 복잡한 AI 워크플로우를 생성한다.

  2. 상태 관리: 워크플로우 내에서 서로 다른 작업 간의 상태를 쉽게 관리하고 공유할 수 있다.

  3. 이벤트 기반 아키텍처: 이벤트 기반 모델을 기반으로 동적이고 반응형 워크플로우를 지원한다.

  4. 유연한 제어 흐름: 조건부 논리, 루프, 분기 등을 워크플로우 내에서 구현할 수 있다.

​
시작하기

OpenAI를 사용해 무작위 도시를 생성하고, 그 도시에 대한 재미있는 사실을 생성하는 간단한 Flow를 만들어보자.

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. 메서드 이름으로 리스닝: 리스닝할 메서드 이름을 문자열로 전달한다. 해당 메서드가 완료되면 리스너 메서드가 트리거된다.

    @listen("generate_city")
def generate_fun_fact(self, random_city):
    # 구현

  2. 메서드 직접 리스닝: 메서드 자체를 전달한다. 해당 메서드가 완료되면 리스너 메서드가 트리거된다.

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

​
Flow 출력 처리

Flow의 출력에 접근하고 처리하는 것은 AI 워크플로를 더 큰 애플리케이션이나 시스템에 통합하는 데 필수적이다. CrewAI Flows는 최종 출력을 가져오고, 중간 결과에 접근하며, Flow의 전반적인 상태를 관리할 수 있는 직관적인 메커니즘을 제공한다.

​
최종 결과 가져오기

Flow를 실행할 때, 마지막으로 완료되는 메서드가 최종 결과를 결정한다. kickoff() 메서드는 이 최종 메서드의 결과를 반환한다.

최종 결과에 접근하는 방법은 다음과 같다:

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)

이 예제에서 second_method가 마지막으로 완료되는 메서드이므로, 이 메서드의 결과가 Flow의 최종 결과가 된다. kickoff() 메서드는 최종 결과를 반환하며, 이 결과는 콘솔에 출력된다.

​
상태 접근 및 업데이트

최종 출력을 가져오는 것 외에도, Flow 내부의 상태에 접근하고 업데이트할 수 있다. 상태는 Flow 내의 다양한 메서드 간에 데이터를 저장하고 공유하는 데 사용된다. Flow가 실행된 후, 상태에 접근하여 실행 중 추가되거나 업데이트된 정보를 가져올 수 있다.

다음은 상태를 업데이트하고 접근하는 방법을 보여주는 예제다:

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)

이 예제에서 first_method와 second_method가 상태를 업데이트한다. Flow가 실행된 후, 최종 상태에 접근하여 이 메서드들이 수행한 업데이트를 확인할 수 있다.

최종 메서드의 출력을 반환하고 상태에 접근할 수 있도록 함으로써, CrewAI Flow는 AI 워크플로우의 결과를 더 큰 애플리케이션이나 시스템에 통합하기 쉽게 한다. 또한 Flow 실행 중 상태를 유지하고 접근할 수 있다.

​
플로우 상태 관리

효과적인 상태 관리는 신뢰할 수 있고 유지보수가 용이한 AI 워크플로우를 구축하는 데 필수적이다. CrewAI Flows는 비정형 및 정형 상태 관리를 위한 강력한 메커니즘을 제공한다. 이를 통해 개발자는 애플리케이션의 요구사항에 가장 적합한 접근 방식을 선택할 수 있다.

​
비정형 상태 관리

비정형 상태 관리에서는 모든 상태가 Flow 클래스의 state 속성에 저장된다. 이 방식은 개발자가 엄격한 스키마를 정의하지 않고도 상태 속성을 동적으로 추가하거나 수정할 수 있는 유연성을 제공한다. 비정형 상태를 사용하더라도 CrewAI Flows는 각 상태 인스턴스에 대해 고유 식별자(UUID)를 자동으로 생성하고 유지한다.

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 시스템에 의해 자동으로 생성되고 관리된다.

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 데코레이터를 클래스 수준에 적용하면 모든 플로우 메서드의 상태가 자동으로 유지된다:

@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를 적용할 수 있다:

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_ 함수는 여러 메서드를 동시에 감시하고, 지정된 메서드 중 하나라도 출력을 내면 리스너 메서드를 실행한다.

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()

이 플로우를 실행하면 logger 메서드는 start_method 또는 second_method 중 하나의 출력에 의해 트리거된다. or_ 함수는 여러 메서드를 감시하고, 지정된 메서드 중 하나라도 출력을 내면 리스너 메서드를 실행하는 데 사용된다.

​
조건부 로직: and

Flows에서 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.kickoff()

이 Flow를 실행하면 logger 메서드는 start_method와 second_method 모두가 출력을 내보낼 때만 트리거된다. and_ 함수는 여러 메서드를 감시하고, 지정된 모든 메서드가 출력을 내보낼 때만 리스너 메서드를 실행하는 데 사용된다.

​
라우터

@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.kickoff()

위 예제에서 start_method는 임의의 불리언 값을 생성해 상태에 설정한다. second_method는 @router() 데코레이터를 사용해 불리언 값에 따라 조건부 라우팅 로직을 정의한다. 불리언 값이 True면 "success"를 반환하고, False면 "failed"를 반환한다. third_method와 fourth_method는 second_method의 출력을 기다렸다가 반환된 값에 따라 실행된다.

이 플로우를 실행하면 start_method에서 생성된 임의의 불리언 값에 따라 출력이 달라진다.

​
플로우에 크루 추가하기

CrewAI에서 여러 크루를 포함한 플로우를 만드는 작업은 간단하다.

다음 커맨드를 실행하면 여러 크루를 포함한 플로우를 만들기 위해 필요한 모든 기본 구조가 포함된 새로운 CrewAI 프로젝트를 생성할 수 있다:

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를 연결하는 예제이다:

#!/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() 메서드를 호출해 시작한다.

​
플로우 실행하기

(선택 사항) 플로우를 실행하기 전에, 다음 명령어를 실행해 필요한 의존성을 설치한다:

crewai install

의존성 설치가 완료되면, 가상 환경을 활성화해야 한다. 다음 명령어를 실행한다:

source .venv/bin/activate

가상 환경을 활성화한 후, 다음 명령어 중 하나를 실행해 플로우를 실행한다:

crewai flow kickoff

또는

uv run kickoff

플로우가 실행되면, 콘솔에서 결과를 확인할 수 있다.

​
플롯 흐름 시각화

AI 워크플로우를 시각화하면 흐름의 구조와 실행 경로를 더 잘 이해할 수 있다. CrewAI는 강력한 시각화 도구를 제공해 여러분의 흐름을 인터랙티브한 플롯으로 생성할 수 있게 한다. 이를 통해 AI 워크플로우를 더 쉽게 이해하고 최적화할 수 있다.

​
플롯이란 무엇인가?

CrewAI에서 플롯은 AI 워크플로우를 시각적으로 표현한 그래픽이다. 각각의 작업, 그들 간의 연결, 그리고 데이터의 흐름을 보여준다. 이 시각화는 작업의 순서를 이해하고, 병목 현상을 식별하며, 워크플로우 로직이 기대한 대로 작동하는지 확인하는 데 도움을 준다.

​
플롯 생성 방법

CrewAI는 여러분의 작업 흐름을 시각화하기 위한 두 가지 편리한 방법을 제공한다:

​
Option 1: plot() 메서드 사용하기

플로우 인스턴스를 직접 다룰 때는 plot() 메서드를 호출해 플롯을 생성할 수 있다. 이 메서드는 플로우의 인터랙티브 플롯을 포함한 HTML 파일을 만든다.

# 플로우 인스턴스가 있다고 가정
flow.plot("my_flow_plot")

이 코드는 현재 디렉토리에 my_flow_plot.html 파일을 생성한다. 웹 브라우저에서 이 파일을 열어 인터랙티브 플롯을 확인할 수 있다.

​
옵션 2: 커맨드라인 사용

구조화된 CrewAI 프로젝트 내에서 작업한다면, 커맨드라인을 통해 플로우 다이어그램을 생성할 수 있다. 이 방법은 전체 플로우 설정을 시각적으로 확인하고 싶은 대규모 프로젝트에 특히 유용하다.

crewai flow plot

이 커맨드를 실행하면 plot() 메서드와 유사한 플로우 다이어그램이 포함된 HTML 파일이 생성된다. 파일은 프로젝트 디렉토리에 저장되며, 웹 브라우저에서 열어 플로우를 탐색할 수 있다.

​
플롯 이해하기

생성된 플롯은 여러분의 흐름에 있는 작업을 나타내는 노드와 실행 흐름을 나타내는 방향성 간선으로 구성된다. 이 플롯은 상호작용이 가능하며, 확대/축소가 가능하고 노드 위에 마우스를 올리면 추가 세부 정보를 확인할 수 있다.

흐름을 시각화하면 워크플로우의 구조를 더 명확히 이해할 수 있다. 이를 통해 디버깅, 최적화, 그리고 AI 프로세스를 다른 사람들에게 설명하기가 더 쉬워진다.

​
결론

CrewAI의 플로우 시각화 기능은 복잡한 AI 워크플로우를 설계하고 관리하는 데 큰 도움이 된다. plot() 메서드를 사용하든 커맨드라인을 통해 플롯을 생성하든, 워크플로우를 시각적으로 표현하면 개발과 프레젠테이션 모두에 유용하다.

​
다음 단계

추가적인 플로우 예제를 탐색하고 싶다면, 예제 저장소에서 다양한 추천 사례를 확인할 수 있다. 여기서는 네 가지 구체적인 플로우 예제를 소개하며, 각각 고유한 사용 사례를 보여준다. 이를 통해 현재 문제 유형에 맞는 예제를 찾는 데 도움을 받을 수 있다.

  1. 이메일 자동 응답 플로우: 이 예제는 백그라운드 작업이 지속적으로 실행되어 이메일 응답을 자동화하는 무한 루프를 보여준다. 수동 개입 없이 반복적으로 수행해야 하는 작업에 적합한 사례이다. 예제 보기

  2. 리드 스코어 플로우: 이 플로우는 인간의 피드백을 포함하고 라우터를 사용해 다양한 조건부 분기를 처리하는 방법을 보여준다. 동적 의사 결정과 인간의 감독을 워크플로우에 통합하는 방법을 배울 수 있는 훌륭한 예제이다. 예제 보기

  3. 책 작성 플로우: 이 예제는 여러 크루를 연결하는 데 탁월하다. 한 크루가 책의 전체 개요를 작성하면, 다른 크루가 그 개요를 바탕으로 각 장을 생성한다. 결국 모든 것이 연결되어 완전한 책을 만든다. 복잡한 다단계 프로세스와 작업 간 조정이 필요한 경우에 적합한 플로우이다. 예제 보기

  4. 회의 어시스턴트 플로우: 이 플로우는 하나의 이벤트를 브로드캐스트하여 여러 후속 작업을 트리거하는 방법을 보여준다. 예를 들어, 회의가 끝난 후 Trello 보드를 업데이트하고, Slack 메시지를 보내고, 결과를 저장하는 등의 작업을 수행할 수 있다. 단일 이벤트에서 여러 결과를 처리하는 방법을 배울 수 있는 훌륭한 예제로, 포괄적인 작업 관리 및 알림 시스템에 적합하다. 예제 보기

이 예제들을 탐색하면 CrewAI 플로우를 다양한 사용 사례에 활용하는 방법을 깊이 있게 이해할 수 있다. 반복 작업 자동화부터 동적 의사 결정과 인간 피드백이 필요한 복잡한 다단계 프로세스 관리까지 다양한 시나리오에 적용할 수 있다.

또한, CrewAI에서 플로우를 사용하는 방법에 대한 YouTube 비디오도 확인해 보자!

CrewsKnowledge