콘텐츠로 이동

첫 플로우 만들기 (First Flow)

CrewAI에서 크루(Crew)가 여러 에이전트가 협력하는 팀이라면, 플로우(Flow)는 그 팀이 언제, 어떤 순서로 일할지를 우리가 직접 조율하는 오케스트레이션 레이어예요. 크루가 에이전트 간 협업에 강점이 있다면, 플로우는 "내 AI 시스템이 어떻게 상호작용할지"를 세밀하게 제어하는 데 강점이 있습니다. 이번 가이드에서는 어떤 주제든 종합 학습 가이드를 생성하는 플로우를 처음부터 만들어 보면서, 플로우가 일반적인 파이썬 코드, 직접적인 LLM 호출, 그리고 크루 기반 처리를 어떻게 하나의 이벤트 중심 시스템으로 엮는지 볼게요.

플로우가 강력한 이유

플로우를 쓰면 이런 일이 가능해집니다.

  1. 서로 다른 AI 상호작용 패턴을 조합 — 복잡한 협업 작업엔 크루, 단순한 작업엔 직접 LLM 호출, 절차적 로직엔 일반 코드를 써요.
  2. 이벤트 중심 시스템 구축 — 특정 이벤트와 데이터 변화에 컴포넌트가 어떻게 반응할지 정의해요.
  3. 컴포넌트 간 상태 유지 — 애플리케이션의 여러 부분 사이에서 데이터를 공유하고 변환해요.
  4. 외부 시스템 통합 — AI 워크플로우를 데이터베이스, API, 사용자 인터페이스와 자연스럽게 연결해요.
  5. 복잡한 실행 경로 설계 — 조건 분기, 병렬 처리, 동적 워크플로우를 만들 수 있어요.

이 가이드를 마치면 사용자 입력과 AI 계획, 다중 에이전트 콘텐츠 생성이 결합된 콘텐츠 생성 시스템을 만들게 됩니다. 각 단계가 이전 단계의 완료에 반응하는 이벤트 중심 구조를 직접 구현해 보면서, 더 복잡한 AI 애플리케이션으로 확장할 바탕을 쌓는 거예요. 같은 패턴으로 대화형 AI 어시스턴트, AI 강화 데이터 처리 파이프라인, 외부 서비스와 통합하는 자율 에이전트, 사람 개입이 들어가는 다단계 의사 결정 시스템까지 만들 수 있습니다.

시작해 볼게요.

준비물

시작 전에 아래가 갖춰져 있는지 확인하세요.

  1. 설치 가이드를 따라 CrewAI를 설치
  2. LLM 설정을 참고해 환경 변수에 LLM API 키 설정
  3. 파이썬 기본 문법 이해

Step 1: CrewAI Flow 프로젝트 만들기

CLI로 새 플로우 프로젝트를 생성합니다. 이 명령이 필요한 디렉터리와 템플릿 파일을 갖춘 스캐폴드를 만들어 줘요.

crewai create flow guide_creator_flow
cd guide_creator_flow

이렇게 하면 플로우를 돌리는 데 필요한 기본 구조가 만들어집니다.

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

생성된 프로젝트는 다음과 같은 구조를 가집니다. 기본으로 들어 있는 크루는 전형적인 Python/YAML 레이아웃을 쓰고, Step 4에서 콘텐츠 크루를 JSONC 크루로 교체할 거예요.

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

이 구조는 플로우의 각 구성 요소를 깔끔하게 분리해 줍니다. 플로우 로직은 src/guide_creator_flow/main.py, 전용 크루는 src/guide_creator_flow/crews, 커스텀 도구는 src/guide_creator_flow/tools에 둬요. 우리는 이 구조를 수정해 종합 학습 가이드를 만드는 플로우로 바꿀 겁니다.

Step 3: 콘텐츠 작성 크루 추가하기

플로우에는 콘텐츠 생성을 맡을 전용 크루가 필요해요. CrewAI CLI로 콘텐츠 크루를 추가할게요.

crewai flow add-crew content-crew

이 명령이 크루에 필요한 디렉터리와 템플릿 파일을 자동으로 만들어 줍니다. 콘텐츠 작성 크루는 가이드의 각 섹션을 쓰고 검토하는 일을 맡고, 전체적인 조율은 메인 애플리케이션인 플로우가 담당해요.

Step 4: 콘텐츠 작성 크루 설정하기 (JSONC)

이제 콘텐츠 크루를 JSONC로 설정할게요. 작가(writer)와 리뷰어(reviewer) 두 에이전트가 협력해서 고품질 콘텐츠를 만들도록 구성해요.

  1. src/guide_creator_flow/crews/content_crew/agents/content_writer.jsonc 생성:
{
  "role": "Educational Content Writer",
  "goal": "Create engaging, informative content that thoroughly explains the assigned topic and provides valuable insights to the reader.",
  "backstory": "You are a talented educational writer who explains complex concepts in accessible language and organizes information clearly.",
  "llm": "provider/model-id",
  "settings": {
    "verbose": true
  }
}
  1. src/guide_creator_flow/crews/content_crew/agents/content_reviewer.jsonc 생성:
{
  "role": "Educational Content Reviewer and Editor",
  "goal": "Ensure content is accurate, comprehensive, well-structured, and consistent with previously written sections.",
  "backstory": "You are a meticulous editor with an eye for detail, clarity, and coherence.",
  "llm": "provider/model-id",
  "settings": {
    "verbose": true
  }
}

provider/model-id는 실제 사용할 모델로 바꿔주세요. 예를 들어 openai/gpt-4o, gemini/gemini-3.7-flash, anthropic/claude-sonnet-4-6처럼요.

  1. src/guide_creator_flow/crews/content_crew/crew.jsonc 생성:
{
  "name": "Content Crew",
  "agents": ["content_writer", "content_reviewer"],
  "tasks": [
    {
      "name": "write_section_task",
      "description": "Write a comprehensive section on the topic: \"{section_title}\".\n\nSection description: {section_description}\nTarget audience: {audience_level} level learners\n\nYour content should begin with a brief introduction, explain key concepts clearly with examples, include practical applications where appropriate, end with a summary, and be approximately 500-800 words.\n\nPreviously written sections:\n{previous_sections}",
      "expected_output": "A well-structured, comprehensive section in Markdown format that thoroughly explains the topic and is appropriate for the target audience.",
      "agent": "content_writer",
      "markdown": true
    },
    {
      "name": "review_section_task",
      "description": "Review and improve this section on \"{section_title}\":\n\n{draft_content}\n\nTarget audience: {audience_level} level learners\nPreviously written sections:\n{previous_sections}\n\nFix errors, improve clarity, verify consistency, enhance structure, and add missing key information.",
      "expected_output": "An improved, polished version of the section that maintains the original structure but enhances clarity, accuracy, and consistency.",
      "agent": "content_reviewer",
      "context": ["write_section_task"],
      "markdown": true
    }
  ],
  "process": "sequential",
  "verbose": true
}

context 필드가 리뷰어가 작가의 출력을 사용할 수 있게 해줍니다.

  1. src/guide_creator_flow/crews/content_crew/content_crew.py를 작은 로더로 교체:
from pathlib import Path

from crewai.project import load_crew

def kickoff_content_crew(inputs: dict):
    crew, default_inputs = load_crew(Path(__file__).with_name("crew.jsonc"))
    return crew.kickoff(inputs={**default_inputs, **inputs})

이 로더가 실행 시점에 crew.jsonc를 하나의 Crew로 변환해요. 이 크루는 단독으로도 동작하지만, 우리 플로우 안에서는 더 큰 시스템의 일부로 오케스트레이션됩니다.

Step 5: 플로우 만들기

이제 재미있는 부분, 즉 가이드 생성 과정 전체를 오케스트레이션하는 플로우를 만들 차례예요. 여기서 일반 파이썬 코드, 직접 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 kickoff_content_crew

# Define our models for structured data
class Section(BaseModel):
    title: str = Field(description="Title of the section")
    description: str = Field(description="Brief description of what the section should cover")

class GuideOutline(BaseModel):
    title: str = Field(description="Title of the guide")
    introduction: str = Field(description="Introduction to the topic")
    target_audience: str = Field(description="Description of the target audience")
    sections: List[Section] = Field(description="List of sections in the guide")
    conclusion: str = Field(description="Conclusion or summary of the guide")

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

class GuideCreatorFlow(Flow[GuideCreatorState]):
    """Flow for creating a comprehensive guide on any topic"""

    @start()
    def get_user_input(self):
        """Get input from the user about the guide topic and audience"""
        print("\n=== Create Your Comprehensive Guide ===\n")

        # Get user input
        self.state.topic = input("What topic would you like to create a guide for? ")

        # Get audience level with validation
        while True:
            audience = input("Who is your target audience? (beginner/intermediate/advanced) ").lower()
            if audience in ["beginner", "intermediate", "advanced"]:
                self.state.audience_level = audience
                break
            print("Please enter 'beginner', 'intermediate', or 'advanced'")

        print(f"\nCreating a guide on {self.state.topic} for {self.state.audience_level} audience...\n")
        return self.state

    @listen(get_user_input)
    def create_guide_outline(self, state):
        """Create a structured outline for the guide using a direct LLM call"""
        print("Creating guide outline...")

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

        # Create the messages for the outline
        messages = [
            {"role": "system", "content": "You are a helpful assistant designed to output JSON."},
            {"role": "user", "content": f"""
            Create a detailed outline for a comprehensive guide on "{state.topic}" for {state.audience_level} level learners.

            The outline should include:
            1. A compelling title for the guide
            2. An introduction to the topic
            3. 4-6 main sections that cover the most important aspects of the topic
            4. A conclusion or summary

            For each section, provide a clear title and a brief description of what it should cover.
            """}
        ]

        # Make the LLM call with JSON response format
        response = llm.call(messages=messages)

        # Parse the JSON response
        outline_dict = json.loads(response)
        self.state.guide_outline = GuideOutline(**outline_dict)

        # Ensure output directory exists before saving
        os.makedirs("output", exist_ok=True)

        # Save the outline to a file
        with open("output/guide_outline.json", "w") as f:
            json.dump(outline_dict, f, indent=2)

        print(f"Guide outline created with {len(self.state.guide_outline.sections)} sections")
        return self.state.guide_outline

    @listen(create_guide_outline)
    def write_and_compile_guide(self, outline):
        """Write all sections and compile the guide"""
        print("Writing guide sections and compiling...")
        completed_sections = []

        # Process sections one by one to maintain context flow
        for section in outline.sections:
            print(f"Processing section: {section.title}")

            # Build context from previous sections
            previous_sections_text = ""
            if completed_sections:
                previous_sections_text = "# Previously Written Sections\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 = "No previous sections written yet."

            # Run the content crew for this section
            result = kickoff_content_crew(inputs={
                "section_title": section.title,
                "section_description": section.description,
                "audience_level": self.state.audience_level,
                "previous_sections": previous_sections_text,
                "draft_content": ""
            })

            # Store the content
            self.state.sections_content[section.title] = result.raw
            completed_sections.append(section.title)
            print(f"Section completed: {section.title}")

        # Compile the final guide
        guide_content = f"# {outline.title}\n\n"
        guide_content += f"## Introduction\n\n{outline.introduction}\n\n"

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

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

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

        print("\nComplete guide compiled and saved to output/complete_guide.md")
        return "Guide creation completed successfully"

def kickoff():
    """Run the guide creator flow"""
    GuideCreatorFlow().kickoff()
    print("\n=== Flow Complete ===")
    print("Your comprehensive guide is ready in the output directory.")
    print("Open output/complete_guide.md to view it.")

def plot():
    """Generate a visualization of the flow"""
    flow = GuideCreatorFlow()
    flow.plot("guide_creator_flow")
    print("Flow visualization saved to guide_creator_flow.html")

if __name__ == "__main__":
    kickoff()

이 플로우에서 어떤 일이 벌어지는지 짚어 볼게요.

  1. 구조화된 데이터를 위한 Pydantic 모델을 정의해 타입 안전성과 명확한 데이터 표현을 보장해요.
  2. 플로우의 여러 단계에 걸쳐 데이터를 유지하는 상태(state) 클래스를 만들어요.
  3. 세 가지 핵심 단계를 구현해요 — @start() 데코레이터로 사용자 입력 받기, 직접 LLM 호출로 개요 만들기, 콘텐츠 크루로 섹션 처리.
  4. @listen() 데코레이터로 단계 간 이벤트 중심 관계를 만들어요.

이게 바로 플로우의 힘입니다. 사용자 상호작용, 직접 LLM 호출, 크루 기반 작업 같은 서로 다른 처리 방식을 하나의 일관된 이벤트 중심 시스템으로 엮는 거예요.

Step 6: 환경 변수 설정

프로젝트 루트에 .env 파일을 만들고 API 키를 넣어주세요. 프로바이더 설정 방법은 LLM 설정 가이드를 참고하면 돼요.

.env

OPENAI_API_KEY=your_openai_api_key
# or
GEMINI_API_KEY=your_gemini_api_key
# or
ANTHROPIC_API_KEY=your_anthropic_api_key

Step 7: 의존성 설치

필요한 의존성을 설치해요.

crewai install

Step 8: 플로우 실행

이제 플로우를 실제로 돌려볼 차례예요. CrewAI CLI로 실행해요.

crewai run

실행하면 플로우가 살아 움직이는 걸 볼 수 있습니다. 주제와 대상 수준을 물어보고, 구조화된 가이드 개요를 만든 뒤, 각 섹션을 콘텐츠 작가와 리뷰어가 협력해 처리하고, 마지막으로 모든 것을 종합 가이드로 합쳐요. 이 과정이 AI 컴포넌트와 비-AI 컴포넌트가 뒤섞인 복잡한 프로세스를 플로우가 어떻게 오케스트레이션하는지 보여줍니다.

Step 9: 플로우 시각화

플로우의 강력한 기능 중 하나는 구조를 시각화할 수 있다는 점이에요.

crewai flow plot

이 명령은 플로우의 구조를 보여주는 HTML 파일을 만듭니다. 단계들 사이의 관계와 단계 사이를 흐르는 데이터까지 표시되죠. 복잡한 플로우를 이해하고 디버깅하는 데 아주 유용해요.

Step 10: 출력물 확인

플로우가 끝나면 output 디렉터리에 두 파일이 생깁니다.

  1. guide_outline.json — 가이드의 구조화된 개요
  2. complete_guide.md — 모든 섹션이 담긴 종합 가이드

시간을 내서 이 파일들을 살펴보세요. 사용자 입력, 직접 AI 상호작용, 에이전트 협업이 결합해 고품질 결과물을 만들어 내는 시스템을 직접 만든 겁니다.

실무 관점: 첫 플로우 너머

이 가이드에서 배운 내용은 훨씬 정교한 AI 시스템을 만들기 위한 바탕이 됩니다. 기본 플로우를 확장할 수 있는 방향을 몇 가지 볼게요.

  • 사용자 상호작용 강화 — 입출력을 위한 웹 인터페이스, 실시간 진행 상황 표시, 피드백을 반영하는 개선 루프, 다단계 사용자 상호작용을 추가할 수 있어요.
  • 처리 단계 추가 — 개요 작성 전에 리서치 단계를 넣거나, 삽화 이미지 생성, 기술 가이드용 코드 스니펫 생성, 최종 품질 검증과 사실 확인 단계를 붙일 수 있어요.
  • 더 복잡한 플로우 패턴 — 사용자 선호나 콘텐츠 유형에 따른 조건 분기, 독립 섹션의 병렬 처리, 피드백이 들어간 반복 개선 루프, 외부 API·서비스 연동까지 가능해요.
  • 다른 도메인에 적용 — 같은 패턴으로 대화형 스토리텔링, 데이터를 처리해 인사이트와 리포트를 만드는 비즈니스 인텔리전스, 아이디어 발굴·설계·계획을 돕는 제품 개발, 맞춤형 학습 경험을 만드는 교육 시스템까지 만들 수 있어요.

이 프로젝트가 보여주는 핵심 기능을 정리하면 이렇습니다.

  1. 사용자 상호작용 — 입력을 사용자에게서 직접 수집
  2. 직접 LLM 호출 — 단일 목적의 효율적인 AI 상호작용을 위해 LLM 클래스 사용
  3. Pydantic 기반 구조화 데이터 — 타입 안전성 보장
  4. 컨텍스트가 있는 순차 처리 — 이전 섹션을 컨텍스트로 제공하며 순서대로 작성
  5. 다중 에이전트 크루 — 콘텐츠 생성을 위해 전문화된 작가와 리뷰어 활용
  6. 상태 관리 — 프로세스의 여러 단계에 걸쳐 상태 유지
  7. 이벤트 중심 아키텍처 — @listen 데코레이터로 이벤트에 반응

더 알아보기

첫 플로우를 만들었으니 이제 이런 것들을 시도해 볼 수 있어요.

  1. 더 복잡한 플로우 구조와 패턴 실험
  2. @router()로 플로우에 조건 분기 만들기
  3. and_와 or_ 함수로 더 복잡한 병렬 실행 다루기
  4. 플로우를 외부 API, 데이터베이스, 사용자 인터페이스에 연결
  5. 하나의 플로우에 여러 전문화 크루 결합
  6. 대화형 플로우(Conversational Flows)로 다중 턴 채팅 앱 만들기 — 메시지마다 kickoff를 호출하고, ChatSession과 지연 트레이싱(deferred tracing)을 활용하는 방식이에요.

축하합니다! 일반 코드, 직접 LLM 호출, 크루 기반 처리를 결합해 종합 가이드를 만드는 첫 플로우를 완성했어요. 이 기초 위에서 절차적 제어와 협업 지능을 조합해 복잡한 다단계 문제를 다루는, 점점 더 정교한 AI 애플리케이션을 만들어 갈 수 있습니다.