첫 번째 Flow 만들기

Flow로 AI 워크플로우 제어하기

CrewAI의 Flows는 AI 오케스트레이션의 다음 단계입니다. Flows는 에이전트 간 협업이라는 Crew의 강점에, 절차적 프로그래밍의 정밀성과 유연성을 더합니다. Crews가 협업에 특화되어 있다면, Flows는 각 구성 요소가 언제 어떻게 상호작용할지를 정밀하게 제어할 수 있도록 합니다.

이 가이드에서는 어떤 주제에 대해서든 학습용 가이드 문서를 생성하는 Flow를 만들어 봅니다. 이 예제를 통해 Flows가 일반 코드, 직접적인 LLM 호출, Crew 기반 처리 등을 조합하여 구조화된 이벤트 기반 AI 워크플로우를 어떻게 구성하는지 알 수 있습니다.

Flows의 강력한 기능

Flows를 사용하면 다음을 할 수 있습니다:

1. 다양한 AI 상호작용 패턴 결합 – 복잡한 협업은 Crew, 단순 작업은 직접 LLM 호출, 로직 제어는 일반 코드로 처리
2. 이벤트 기반 시스템 구축 – 데이터 변화나 작업 완료 시 자동으로 다음 작업 수행
3. 구성요소 간 상태 유지 – 전체 시스템에 걸쳐 데이터 공유 및 변환
4. 외부 시스템과 통합 – 데이터베이스, API, UI 등과 유기적으로 연동 가능
5. 복잡한 실행 경로 설계 – 조건 분기, 병렬 처리, 동적 워크플로우 설계 가능

이 가이드를 통해 배우는 것

이 가이드를 마치면 다음을 달성할 수 있습니다:

1. 사용자 입력, AI 기획, 에이전트 협업 기반 콘텐츠 생성 시스템 구현
2. 시스템 구성요소 간 정보 흐름 오케스트레이션
3. 이벤트 기반 아키텍처 구현 – 각 단계가 앞선 단계의 완료에 반응
4. 확장 가능한 복잡한 AI 애플리케이션 설계 기초 확보

이 가이드 생성 Flow는 다음과 같은 고급 응용에도 적용 가능한 기본 패턴을 보여줍니다:

• 다중 하위 시스템이 협업하는 인터랙티브 AI 비서
• AI 기반 변환 작업이 포함된 데이터 처리 파이프라인
• 외부 API 및 서비스와 통합된 자율 에이전트
• 사용자 개입 기반 다단계 의사결정 시스템

사전 준비 사항

시작하기 전에 다음을 준비하세요:

1. 설치 가이드를 참고하여 CrewAI 설치
2. 환경 변수에 OpenAI API 키 설정
3. Python에 대한 기본 이해

1단계: 새로운 Flow 프로젝트 생성

CLI를 통해 새로운 Flow 프로젝트를 생성합니다. 아래 명령어는 Flow에 필요한 디렉터리 및 템플릿 파일을 자동으로 생성해줍니다:

crewai create flow guide_creator_flow
cd guide_creator_flow

이렇게 하면 Flow 구현에 필요한 기본 구조가 자동으로 생성됩니다.

2단계: 프로젝트 구조 이해하기

생성된 프로젝트의 구조는 다음과 같습니다. 이 구조를 이해해두면 더 복잡한 Flow를 설계할 때 큰 도움이 됩니다.

guide_creator_flow/
├── .gitignore
├── pyproject.toml
├── README.md
├── .env
├── main.py
├── crews/
│   └── poem_crew/
│       ├── config/
│       │   ├── agents.yaml
│       │   └── tasks.yaml
│       └── poem_crew.py
└── tools/
    └── custom_tool.py

구성요소 별 역할은 다음과 같습니다:

• main.py: Flow의 핵심 실행 로직
• crews/: 각각의 Crew 정의 디렉터리
• tools/: 사용자 정의 도구 모음 디렉터리

이제 이 구조를 바탕으로 학습용 가이드를 만드는 Flow를 구현해보겠습니다.

3단계: 콘텐츠 작성 Crew 추가하기

이 Flow에는 콘텐츠를 생성할 전문 Crew가 필요합니다. 다음 명령어로 콘텐츠 작성 Crew를 추가하세요:

crewai flow add-crew content-crew

이 명령어는 콘텐츠 작성 Crew를 위한 디렉터리와 기본 파일들을 자동 생성합니다. 이 Crew는 가이드의 각 섹션을 작성하고 검토하는 역할을 하게 되며, 전체 Flow의 일부로서 작동합니다.

4단계: 콘텐츠 작성 크루 설정하기

이제 콘텐츠 작성 크루에 필요한 파일을 수정해보겠습니다. 이 크루에는 작문과 리뷰를 담당할 두 명의 전문 에이전트를 설정합니다.

1. 에이전트 구성 파일 설정

먼저 agents.yaml 파일을 아래와 같이 수정해 팀을 구성합니다:

# src/guide_creator_flow/crews/content_crew/config/agents.yaml
content_writer:
  role: >
    교육 콘텐츠 작가
  goal: >
    할당된 주제를 철저하게 설명하고 독자에게 유용한 인사이트를 제공하는 흥미롭고 유익한 콘텐츠를 작성합니다.
  backstory: >
    당신은 복잡한 개념을 쉽게 설명하고 정보를 체계적으로 정리하는 데 능숙한 교육 콘텐츠 전문가입니다.
  llm: openai/gpt-4o-mini

content_reviewer:
  role: >
    교육 콘텐츠 리뷰어 및 편집자
  goal: >
    콘텐츠의 정확성, 포괄성, 구조적 완성도, 이전 섹션과의 일관성을 확보합니다.
  backstory: >
    수년간 교육 콘텐츠를 리뷰한 경험이 있으며, 명확성과 일관성을 중시하는 꼼꼼한 편집자입니다. 원작자의 문체를 유지하면서 품질을 향상시키는 데 능숙합니다.
  llm: openai/gpt-4o-mini

각 에이전트는 고유한 목적과 전문성을 갖고 있으며, 이 설정은 에이전트의 작업 방식에 영향을 줍니다.

2. 태스크 구성 파일 설정

다음으로 tasks.yaml 파일을 수정해 각 에이전트가 수행할 작업을 정의합니다:

# src/guide_creator_flow/crews/content_crew/config/tasks.yaml
write_section_task:
  description: >
    다음 주제에 대한 종합적인 섹션을 작성하세요: "{section_title}"

    섹션 설명: {section_description}
    대상 학습 수준: {audience_level}

    작성 지침:
    1. 간단한 소개로 시작하세요
    2. 핵심 개념을 명확히 설명하고 예시를 제공하세요
    3. 실용적 적용 사례나 연습을 포함하세요
    4. 핵심 요점을 요약하면서 마무리하세요
    5. 분량은 약 500~800단어로 구성하세요

    Markdown 형식으로 작성하고 제목, 리스트, 강조를 적절히 사용하세요.

    이전에 작성된 섹션:
    {previous_sections}

    이전 섹션과의 일관성을 유지하며, 이미 설명된 개념 위에 구축되도록 하세요.
  expected_output: >
    주제를 철저히 설명하고 대상 독자 수준에 맞춘 Markdown 형식의 구조화된 콘텐츠 섹션
  agent: content_writer

review_section_task:
  description: >
    다음 섹션에 대해 리뷰하고 개선하세요: "{section_title}"

    {draft_content}

    대상 학습 수준: {audience_level}

    이전에 작성된 섹션:
    {previous_sections}

    리뷰 지침:
    1. 문법 및 철자 오류 수정
    2. 명확성과 가독성 향상
    3. 내용의 포괄성과 정확성 검토
    4. 이전 섹션과의 일관성 확인
    5. 구조 및 흐름 개선
    6. 누락된 주요 정보 추가

    개선된 Markdown 형식의 콘텐츠를 제공하세요.
  expected_output: >
    원래 구조를 유지하면서 명확성, 정확성, 일관성을 향상시킨 세련된 섹션
  agent: content_reviewer
  context:
    - write_section_task

이 설정은 작성자가 작성한 초안을 리뷰어가 개선하는 순차적 워크플로우를 구성합니다. context 항목은 리뷰 작업이 작성자의 출력 결과를 사용할 수 있도록 연결합니다.

3. 크루 구현 파일 설정

이제 실제 작동하는 크루를 content_crew.py 파일에 정의합니다:

# src/guide_creator_flow/crews/content_crew/content_crew.py
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List

@CrewBase
class ContentCrew():
    """콘텐츠 작성을 위한 크루"""

    agents: List[BaseAgent]
    tasks: List[Task]

    @agent
    def content_writer(self) -> Agent:
        return Agent(
            config=self.agents_config['content_writer'], # type: ignore[index]
            verbose=True
        )

    @agent
    def content_reviewer(self) -> Agent:
        return Agent(
            config=self.agents_config['content_reviewer'], # type: ignore[index]
            verbose=True
        )

    @task
    def write_section_task(self) -> Task:
        return Task(
            config=self.tasks_config['write_section_task'] # type: ignore[index]
        )

    @task
    def review_section_task(self) -> Task:
        return Task(
            config=self.tasks_config['review_section_task'], # type: ignore[index]
            context=[self.write_section_task()]
        )

    @crew
    def crew(self) -> Crew:
        """콘텐츠 작성 크루 생성"""
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

이 코드는 작성자 → 리뷰어 순으로 작업이 진행되는 구조를 설정합니다. 이 크루는 단독으로도 작동하지만, 전체 Flow 시스템 안에서 조정되어 더욱 큰 프로세스의 일부로 기능하게 됩니다.

5단계: 플로우(Flow) 만들기

이 단계에서는 전체 가이드 생성 과정을 조율할 플로우를 만듭니다다. 여기서는 일반 Python 코드, 직접적인 LLM 호출, 그리고 콘텐츠 제작 크루를 하나의 통합된 시스템으로 결합할 것입니다.

우리가 만들 플로우는 다음과 같은 작업을 수행합니다:

1. 주제와 대상 학습 수준에 대한 사용자 입력을 받습니다.
2. 직접적인 LLM 호출을 통해 구조화된 가이드 개요를 생성합니다.
3. 콘텐츠 작성 크루를 사용하여 각 섹션을 순차적으로 처리합니다.
4. 모든 내용을 종합하여 최종적인 포괄적 문서를 완성합니다.

이제 main.py 파일에 플로우를 생성해봅시다:

#!/usr/bin/env python
import json
import os
from typing import List, Dict
from pydantic import BaseModel, Field
from crewai import LLM
from crewai.flow.flow import Flow, listen, start
from guide_creator_flow.crews.content_crew.content_crew import ContentCrew

# 구조화된 데이터를 위한 모델 정의
class Section(BaseModel):
    title: str = Field(description="섹션 제목")
    description: str = Field(description="섹션에서 다룰 내용의 간단한 설명")

class GuideOutline(BaseModel):
    title: str = Field(description="가이드 제목")
    introduction: str = Field(description="주제 소개")
    target_audience: str = Field(description="대상 독자 설명")
    sections: List[Section] = Field(description="가이드의 섹션 목록")
    conclusion: str = Field(description="결론 또는 요약")

# 플로우 상태 정의
class GuideCreatorState(BaseModel):
    topic: str = ""
    audience_level: str = ""
    guide_outline: GuideOutline = None
    sections_content: Dict[str, str] = {}

class GuideCreatorFlow(Flow[GuideCreatorState]):
    """어떤 주제든 포괄적인 가이드를 생성하는 플로우"""

    @start()
    def get_user_input(self):
        """가이드 주제 및 대상 독자에 대한 사용자 입력 받기"""
        print("\n=== 포괄적인 가이드 생성 시작 ===\n")

        # 주제 입력 받기
        self.state.topic = input("어떤 주제에 대한 가이드를 만들고 싶으신가요? ")

        # 대상 독자 입력 받기 (유효성 검사 포함)
        while True:
            audience = input("대상 독자는 누구인가요? (beginner/intermediate/advanced): ").lower()
            if audience in ["beginner", "intermediate", "advanced"]:
                self.state.audience_level = audience
                break
            print("'beginner', 'intermediate', 또는 'advanced' 중 하나를 입력해주세요.")

        print(f"\n{self.state.audience_level} 수준의 독자를 위한 '{self.state.topic}' 가이드를 생성합니다...\n")
        return self.state

    @listen(get_user_input)
    def create_guide_outline(self, state):
        """직접 LLM 호출을 통해 가이드 개요 생성"""
        print("가이드 개요 생성 중...")

        llm = LLM(model="openai/gpt-4o-mini", response_format=GuideOutline)

        messages = [
            {"role": "system", "content": "You are a helpful assistant designed to output JSON."},
            {"role": "user", "content": f"""
            "{state.topic}"에 대한 포괄적인 가이드의 상세 개요를 생성해주세요.
            대상 독자 수준: {state.audience_level}

            개요에는 다음이 포함되어야 합니다:
            1. 매력적인 가이드 제목
            2. 주제에 대한 소개
            3. 주제를 다각도로 설명하는 4~6개의 주요 섹션
            4. 결론 또는 요약

            각 섹션에는 명확한 제목과 간단한 설명을 포함해주세요.
            """}
        ]

        response = llm.call(messages=messages)
        outline_dict = json.loads(response)
        self.state.guide_outline = GuideOutline(**outline_dict)

        os.makedirs("output", exist_ok=True)
        with open("output/guide_outline.json", "w") as f:
            json.dump(outline_dict, f, indent=2)

        print(f"{len(self.state.guide_outline.sections)}개의 섹션이 포함된 가이드 개요가 생성되었습니다.")
        return self.state.guide_outline

    @listen(create_guide_outline)
    def write_and_compile_guide(self, outline):
        """모든 섹션을 작성하고 가이드를 컴파일합니다"""
        print("가이드 섹션 작성 및 컴파일 중...")
        completed_sections = []

        for section in outline.sections:
            print(f"섹션 처리 중: {section.title}")

            previous_sections_text = ""
            if completed_sections:
                previous_sections_text = "# 이전 섹션 내용\n\n"
                for title in completed_sections:
                    previous_sections_text += f"## {title}\n\n"
                    previous_sections_text += self.state.sections_content.get(title, "") + "\n\n"
            else:
                previous_sections_text = "이전 섹션이 없습니다."

            result = ContentCrew().crew().kickoff(inputs={
                "section_title": section.title,
                "section_description": section.description,
                "audience_level": self.state.audience_level,
                "previous_sections": previous_sections_text,
                "draft_content": ""
            })

            self.state.sections_content[section.title] = result.raw
            completed_sections.append(section.title)
            print(f"섹션 완료: {section.title}")

        guide_content = f"# {outline.title}\n\n"
        guide_content += f"## Introduction\n\n{outline.introduction}\n\n"

        for section in outline.sections:
            section_content = self.state.sections_content.get(section.title, "")
            guide_content += f"\n\n{section_content}\n\n"

        guide_content += f"## Conclusion\n\n{outline.conclusion}\n\n"

        with open("output/complete_guide.md", "w") as f:
            f.write(guide_content)

        print("\n최종 가이드가 output/complete_guide.md에 저장되었습니다.")
        return "가이드 생성이 성공적으로 완료되었습니다."

def kickoff():
    """가이드 생성 플로우 실행"""
    GuideCreatorFlow().kickoff()
    print("\n=== 플로우 완료 ===")
    print("output 디렉토리에서 완성된 가이드를 확인하세요.")
    print("output/complete_guide.md 파일을 열어보세요.")

def plot():
    """플로우 구조 시각화"""
    flow = GuideCreatorFlow()
    flow.plot("guide_creator_flow")
    print("플로우 시각화가 guide_creator_flow.html에 저장되었습니다.")

if __name__ == "__main__":
    kickoff()

이 플로우에서 어떤 일이 일어나는지 분석해봅시다:

1. 구조화된 데이터를 위한 Pydantic 모델을 정의하여 타입 안정성과 명확한 데이터 표현을 보장합니다.
2. 플로우의 여러 단계 간 데이터를 유지하기 위해 상태(state) 클래스를 생성합니다.
3. 세 가지 주요 플로우 단계를 구현합니다:
   • @start() 데코레이터를 사용해 사용자 입력을 받습니다
   • 직접적인 LLM 호출로 가이드 개요를 생성합니다
   • 콘텐츠 크루를 사용해 각 섹션을 처리합니다
4. @listen() 데코레이터를 활용해 단계 간 이벤트 기반 관계를 설정합니다

이것으로 사용자 상호작용, LLM 호출, 크루 기반 작업 등 다양한 처리 방식을 하나의 일관된 이벤트 기반 시스템으로 결합할 수 있습니다.

6단계: 환경 변수 설정하기

프로젝트 루트 디렉터리에 .env 파일을 생성하고 아래와 같이 API 키를 입력하세요:

OPENAI_API_KEY=your_openai_api_key

7단계: 의존성 설치하기

다음 명령어를 통해 필요한 의존성을 설치하세요:

crewai install

8단계: 플로우 실행하기

이제 직접 플로우를 실행해볼 차례입니다! 아래 명령어를 통해 실행할 수 있습니다:

crewai flow kickoff

이 명령어를 실행하면 플로우가 다음 과정을 따릅니다:

1. 주제와 대상 학습 수준에 대한 사용자 입력 요청
2. 주제에 대한 구조화된 가이드 개요 생성
3. 각 섹션을 콘텐츠 작성 및 검토 크루가 순차적으로 처리
4. 모든 내용을 종합하여 완성된 가이드 문서 생성

이를 통해 CrewAI Flows가 AI 및 일반 로직을 포함한 복잡한 프로세스를 조율하는 능력을 직접 체험할 수 있습니다.

9단계: 플로우 시각화하기

CrewAI의 강력한 기능 중 하나는 플로우의 구조를 시각적으로 확인할 수 있다는 것입니다. 다음 명령어를 입력하세요:

crewai flow plot

이 명령어는 각 단계의 관계와 데이터 흐름을 보여주는 HTML 파일을 생성합니다. 복잡한 플로우를 이해하고 디버깅하는 데 유용합니다.

10단계: 출력 결과 확인하기

플로우 실행이 완료되면 output 디렉토리에 다음과 같은 결과물이 생성됩니다:

1. guide_outline.json: 구조화된 가이드 개요
2. complete_guide.md: 전체 섹션이 포함된 최종 가이드 문서

이 파일들을 검토하며, 사용자 입력과 AI 협업이 결합되어 고품질의 복합 콘텐츠를 생성해낸 결과를 확인해보세요.

가능성의 확장: 더 나아가기

이 가이드를 통해 배운 내용은 더 복잡하고 강력한 AI 시스템을 구축하는 기반이 됩니다. 다음과 같은 확장이 가능합니다:

사용자 상호작용 강화

• 웹 인터페이스를 통한 입력/출력
• 실시간 진행 상태 업데이트
• 피드백 기반 반복 개선 루프
• 다단계 사용자 입력 흐름

처리 단계 추가

• 개요 생성을 위한 사전 리서치
• 일러스트용 이미지 생성
• 기술 가이드용 코드 스니펫 생성
• 품질 검증 및 사실 확인 단계 추가

복잡한 플로우 패턴 구현

• 사용자 설정 또는 콘텐츠 유형에 따른 조건 분기
• 독립적인 섹션 병렬 처리
• 피드백을 통한 반복 개선 루프
• 외부 API 또는 서비스와의 통합

다양한 도메인에 적용

• 인터랙티브 스토리텔링: 사용자 입력을 기반으로 맞춤형 이야기 생성
• 비즈니스 인텔리전스: 데이터 분석 및 인사이트 리포트 생성
• 제품 기획: 아이디어 구상부터 설계, 계획까지 조율
• 교육 시스템: 개인 맞춤형 학습 콘텐츠 제공

이 플로우가 보여준 주요 기능

1. 사용자 입력 처리: 플로우 시작 시 직접 사용자로부터 정보를 수집
2. 직접 LLM 호출: 단일 목적의 구조화된 응답 생성을 위해 LLM 클래스 사용
3. Pydantic을 통한 구조화 데이터 처리: 타입 안정성과 명확한 데이터 표현 보장
4. 순차적 처리와 컨텍스트 전달: 앞선 섹션 내용을 다음 단계에 제공하며 순차적으로 처리
5. 멀티에이전트 크루: 전문화된 작가 및 검토자가 협력하여 콘텐츠 생성
6. 상태 관리: 여러 단계에 걸쳐 데이터를 유지하고 전달
7. 이벤트 기반 아키텍처: @listen 데코레이터를 활용하여 단계 간 흐름 연결

플로우 구조 이해하기

1. 직접 LLM 호출

단순하고 구조화된 응답이 필요할 때는 다음과 같이 직접 LLM을 호출할 수 있습니다:

llm = LLM(model="openai/gpt-4o-mini", response_format=GuideOutline)
response = llm.call(messages=messages)

복잡한 협업이 아닌 단일 목적의 출력이 필요할 경우, 크루 대신 이 방법이 효율적입니다.

2. 이벤트 기반 아키텍처

Flow에서는 데코레이터를 사용해 구성 요소 간의 관계를 정의합니다:

@start()
def get_user_input(self):
    # 플로우의 첫 번째 단계
    # ...

@listen(get_user_input)
def create_guide_outline(self, state):
    # get_user_input 단계가 완료되었을 때 실행됨
    # ...

이 방식은 애플리케이션의 구조를 명확하고 선언적으로 만들어줍니다.

3. 상태 관리 (State Management)

Flow는 단계별로 상태(state)를 유지하여 데이터 공유를 쉽게 만들어줍니다:

class GuideCreatorState(BaseModel):
    topic: str = ""
    audience_level: str = ""
    guide_outline: GuideOutline = None
    sections_content: Dict[str, str] = {}

이 구조는 전체 흐름에서 데이터의 추적 및 변환을 타입 안전하게 관리할 수 있게 해줍니다.

4. 크루 통합 (Crew Integration)

Flow는 복잡한 협업 작업을 위해 크루와도 자연스럽게 통합할 수 있습니다:

result = ContentCrew().crew().kickoff(inputs={
    "section_title": section.title,
    # ...
})

이렇게 하면, 애플리케이션의 각 파트에 가장 적합한 도구를 사용할 수 있습니다. 간단한 작업에는 직접 LLM 호출을, 복잡한 협업에는 크루를 사용할 수 있습니다.