CrewAI Flow 간단한 소개

이번에 노마드코더의 “AI Agents 마스터클래스” 라는 강의를 수강하게 되었다. 내용에 비해 가격이 비싸서 다시는 안듣겠다고 다짐했지만, 또 이런 강의를 찾기가 어려운 상황이라 선택의 여지가 없었다. 10월 중순에 사내에서 열리는 AI 활용관련 해커톤에도 활용할 수 있을것 같다는 생각에 이 강의를 지르고야 말았다. ㅎㅎ

각설하고 CrewAI Flow는 이번에 처음 알게 되어 다음과 같이 간단히 클로드를 시켜서 정리해 보았다.

CrewAI란?

CrewAI는 여러 AI 에이전트가 팀처럼 협력해서 복잡한 작업을 수행하도록 하는 멀티 에이전트 오케스트레이션 프레임워크다. João Moura가 2023년 11월에 만든 오픈소스 프로젝트로, 각 에이전트가 특정 역할을 맡아 자율적으로 작업하면서도 하나의 목표를 향해 협력하는 구조를 제공한다. 공식 사이트 주소: https://docs.crewai.com/en/concepts/flows

왜 CrewAI가 만들어졌을까?

CrewAI는 2023년 11월에 출시되어 팀 협업 중심의 멀티 에이전트 오케스트레이션을 제시했다. 이후 2024년 1월 LangChain에서 LangGraph를 출시하면서 멀티 에이전트 프레임워크 경쟁이 본격화되었다. 두 프레임워크는 서로 다른 철학으로 멀티 에이전트 시스템 구축 문제에 접근한다.

LangGraph vs CrewAI 핵심 차이

• LangGraph: 그래프 기반 워크플로우. 노드와 엣지로 복잡한 흐름 제어, 상태 관리, 분기/루프 처리에 강점
• CrewAI: 팀 중심 오케스트레이션. 역할 기반으로 에이전트를 배치하고 협업 구조 구성에 강점

CrewAI는 AutoGen의 유연한 대화형 에이전트와 ChatDev의 구조화된 프로세스를 결합한 접근 방식으로 설계되었으며, 특히 개발 환경에서의 접근성을 강조한다.

Flow: CrewAI의 새로운 차원

Flow는 CrewAI에서 2024년 도입한 기능으로, 단순한 에이전트 협업을 넘어 전체 AI 워크플로우를 제어할 수 있게 해준다. Flow를 통해 일반 Python 코드, LLM 직접 호출, Crew 기반 처리를 하나의 이벤트 기반 시스템으로 통합할 수 있다.

Flow의 핵심 장점

• 다양한 AI 패턴 결합: 복잡한 작업엔 Crew, 단순한 작업엔 직접 LLM 호출, 로직 처리엔 일반 코드 사용
• 이벤트 기반 시스템: 특정 이벤트나 데이터 변화에 반응하도록 구성 가능
• 상태 관리: 워크플로우 전체에 걸쳐 데이터를 공유하고 추적
• 복잡한 실행 경로: 조건부 분기, 병렬 처리, 동적 워크플로우 구현

예제로 이해하는 CrewAI Flow

다음 코드는 Flow의 기본 개념을 보여주는 예제다. 하나씩 살펴보자.

1from crewai.flow.flow import Flow, listen, start, router, and_, or_2from pydantic import BaseModel3 4 5class MyFirstFlowState(BaseModel):6 7    user_id: int = 18    is_admin: bool = False9 10 11class MyFirstFlow(Flow[MyFirstFlowState]):12 13    @start()14    def first(self):15        print(self.state.user_id)16        print("hello")17 18    @listen(first)19    def second(self):20        self.state.user_id = 221        print("world")22 23    @listen(first)24    def third(self):25        print("!")26 27    @listen(and_(second, third))28    def final(self):29        print(":)")30 31    @router(final)32    def route(self):33        if self.state.is_admin:34            return "even"35        else:36            return "odd"37 38    @listen("even")39    def handle_even(self):40        print("even")41 42    @listen("odd")43    def handle_odd(self):44        print("odd")45 46 47flow = MyFirstFlow()48 49flow.plot()50flow.kickoff()

1. State 정의: 데이터 저장소

1class MyFirstFlowState(BaseModel):2    user_id: int = 13    is_admin: bool = False

Flow 전체에서 공유되는 상태(State)다. Pydantic 모델을 사용해 타입 안전성을 보장하며, 각 단계에서 이 상태를 읽고 수정할 수 있다.

2. Flow 클래스: 워크플로우의 본체

1class MyFirstFlow(Flow[MyFirstFlowState]):

Flow는 제네릭 클래스로, 앞서 정의한 MyFirstFlowState를 상태로 사용한다. 이 클래스 안에 모든 워크플로우 단계가 메서드로 정의된다.

3. @start() 데코레이터: 시작점

1@start()2def first(self):3    print(self.state.user_id)  # 14    print("hello")

@start()는 Flow의 진입점이다. 워크플로우가 시작되면 이 메서드가 가장 먼저 실행된다. self.state를 통해 현재 상태에 접근할 수 있다.

4. @listen() 데코레이터: 이벤트 기반 실행

1@listen(first)2def second(self):3    self.state.user_id = 24    print("world")5 6@listen(first)7def third(self):8    print("!")

@listen()은 특정 메서드가 완료되면 실행된다. 여기서 second와 third는 모두 first 완료를 듣고(listen) 있으므로 병렬로 실행된다.

실행 순서:

1. first 실행 → “hello” 출력
2. second와 third 병렬 실행 → “world”, ”!” 출력

5. and_() 함수: 모두 완료 대기

1@listen(and_(second, third))2def final(self):3    print(":)")

and_(second, third)는 second와 third 모두 완료될 때까지 기다린다. 두 메서드가 모두 끝나야 final이 실행된다.

6. @router() 데코레이터: 조건부 분기

1@router(final)2def route(self):3    if self.state.is_admin:4        return "even"5    else:6        return "odd"

@router()는 조건에 따라 다음 실행 경로를 결정한다. 여기서는 final 완료 후 is_admin 값에 따라 “even” 또는 “odd” 경로로 분기한다.

7. 분기된 경로 처리

1@listen("even")2def handle_even(self):3    print("even")4 5@listen("odd")6def handle_odd(self):7    print("odd")

라우터가 반환한 문자열 “even” 또는 “odd”를 듣는 메서드다. 초기 상태에서 is_admin=False이므로 handle_odd가 실행된다.

전체 실행 흐름

1시작: first() → "hello" 출력2  ↓3병렬: second() → user_id를 2로 변경, "world" 출력4      third() → "!" 출력5  ↓6대기: and_(second, third) 완료까지 대기7  ↓8실행: final() → ":)" 출력9  ↓10분기: route() → is_admin=False이므로 "odd" 반환11  ↓12실행: handle_odd() → "odd" 출력

8. Flow 실행과 시각화

1flow = MyFirstFlow()2flow.plot()      # Flow 구조를 HTML로 시각화3flow.kickoff()   # Flow 실행

plot(): Flow의 구조를 시각적으로 확인할 수 있는 HTML 파일 생성시킨다. 아래 캡쳐화면을 참고하자.

생성된 html은 아래와 같이 화면에 표시된다.

생성된 파일 실행결과:

kickoff(): Flow가 실행된 결과는 다음과 같다.

Flow가 실행결과 출력:

1hello2world3!4:)5odd

CrewAI의 핵심 개념 정리

Agent (에이전트)

특정 역할을 맡은 AI 작업자다. 각 에이전트는 역할(role), 목표(goal), 배경 스토리(backstory)를 가지며, 지정된 도구(tool)를 사용해 작업을 수행한다.

Task (작업)

에이전트가 수행해야 할 구체적인 작업이다. 작업 설명, 기대 결과, 담당 에이전트를 정의한다.

Crew (크루)

여러 에이전트와 작업을 묶은 팀이다. 순차적(Sequential) 또는 계층적(Hierarchical) 프로세스로 작업을 실행한다.

Flow (플로우)

Crew와 일반 코드, LLM 호출을 조합하여 복잡한 워크플로우를 만든다. 이벤트 기반으로 동작하며 상태를 관리한다.

Tool (도구)

에이전트가 사용하는 기능이다. 웹 검색, 파일 읽기, API 호출 등 LangChain의 도구를 그대로 사용할 수 있다.

or_() 함수: 하나만 완료되면 실행

예제에는 없지만 or_() 함수도 있다:

1@listen(or_(task_a, task_b))2def next_step(self):3    # task_a 또는 task_b 중 하나만 완료되어도 실행4    pass

and_()는 모든 작업 완료를 기다리지만, or_()는 하나만 완료되어도 다음 단계로 진행한다.

CrewAI를 선택해야 하는 경우

• 여러 AI 에이전트가 역할을 나눠 협업하는 구조가 필요할 때
• 팀처럼 작동하는 AI 시스템을 빠르게 구축하고 싶을 때
• LangGraph보다 간단하고 직관적인 API를 원할 때
• 에이전트 간 역할과 책임을 명확히 분리하고 싶을 때

실무 활용 예시

CrewAI Flow는 다음과 같은 실무 작업에 활용된다:

• 콘텐츠 생성 파이프라인: 조사 에이전트 → 작성 에이전트 → 검토 에이전트
• 고객 지원 시스템: 문의 분류 → 정보 검색 → 답변 생성 → 품질 검증
• 데이터 분석 워크플로우: 데이터 수집 → 분석 → 리포트 생성
• 이메일 자동화: 이메일 필터링 → 중요 이메일 분석 → 초안 작성

---

참고 자료

1. CrewAI 공식 문서 - 첫 Flow 빌드하기 (한국어)
2. IBM - crewAI란 무엇인가요?
3. SK DEVOCEAN - CrewAI로 나만의 AI Agent 만들기
4. TrueFoundry Blog - Crewai vs LangGraph: Know The Differences
5. ZenML Blog - LangGraph vs CrewAI: Let’s Learn About the Differences
6. nuvi Blog - Choosing the Right AI Agent Framework: LangGraph vs CrewAI vs OpenAI Swarm
7. Wikibook Brunch - LangGraph로 복잡한 AI 워크플로 구현하기
8. 모두의연구소 블로그 - LangGraph(랭그래프): AI 워크플로우의 혁신, 무엇을 만들 수 있을까?
9. 교보DTS 기술 블로그 - AI Agent 개발의 핵심: RAG와 LangChain, LangGraph 비교 분석
10. 한국정보통신학회논문지 - LangGraph 기반 CrewAI 다중 에이전트 통합 아키텍처