CrewAI 의사결정

결정 프레임워크 이해하기

CrewAI로 애플리케이션을 만들 때 가장 중요한 결정 중 하나는 어떤 구조를 사용할 것인가 입니다. 단일 Crew? Flow? 둘의 조합? 이 가이드는 요구사항을 분석하고 최적의 아키텍처를 설계하는 데 도움을 줍니다.

결정의 핵심은 복잡도(Complexity) 와 정밀도(Precision) 의 관계를 이해하는 것입니다.

이 매트릭스를 통해 다양한 요구사항에 따라 어떤 접근 방식이 적절한지 시각적으로 확인할 수 있습니다.

복잡도-정밀도 매트릭스 해설

복잡도란?

CrewAI에서 복잡도는 다음을 의미합니다:

• 수행해야 할 작업 단계 수
• 다양한 종류의 작업 포함 여부
• 구성 요소 간의 상호의존성
• 조건 분기 로직의 유무
• 전체 워크플로우의 정교함

정밀도란?

정밀도는 다음과 같은 요소를 포함합니다:

• 최종 출력의 정확성 요구 수준
• 구조화된 결과의 필요성
• 반복 실행 시 동일한 결과가 나오는 재현성
• 각 단계에 대한 제어 필요성
• 출력 결과의 허용 오차 범위

4가지 사분면 분석

1. 낮은 복잡도, 낮은 정밀도

특징:

• 간단한 작업, 출력 변동 허용 가능
• 단계 수가 적고, 창의적/탐색적인 목적

추천 구조: 최소한의 에이전트를 가진 단순한 Crew

예시:

• 아이디어 브레인스토밍
• 창의적인 글쓰기 보조
• 간단한 콘텐츠 생성 및 요약

2. 낮은 복잡도, 높은 정밀도

특징:

• 간단하지만 결과의 정확성과 구조가 중요함
• 데이터 처리, 폼 작성 등에 활용

추천 구조: Flows 또는 구조화된 출력의 단순 Crew

예시:

• 데이터 추출/변환
• 폼 자동 작성 및 검증
• JSON/XML 생성
• 간단한 분류 작업

3. 높은 복잡도, 낮은 정밀도

특징:

• 단계가 많고, 탐색적/창의적 결과 허용
• 구성 요소 간의 복잡한 상호작용 포함

추천 구조: 다수의 특화된 에이전트를 포함한 복합 Crew

예시:

• 리서치 및 분석 파이프라인
• 콘텐츠 제작 파이프라인
• 탐색적 데이터 분석

4. 높은 복잡도, 높은 정밀도

특징:

• 다단계 복잡 워크플로우 + 높은 정확성 요구
• 정밀한 처리 및 정확한 출력이 핵심
• 보통 기업용 또는 미션 크리티컬 시스템에 해당

추천 구조: 여러 Crew를 조율하는 Flow, 검증 단계 포함

예시:

• 기업용 의사결정 지원 시스템
• 복잡한 문서 처리 파이프라인
• 규제 산업용 애플리케이션

Crews를 선택할 때

Crews는 다음과 같은 경우에 적합합니다:

1. 협업 지능이 필요한 경우 – 서로 다른 전문성을 가진 여러 에이전트가 함께 일해야 할 때
2. 창의적 해결 방식이 필요한 문제 – 다양한 관점과 접근 방식이 유리한 상황
3. 주로 창의적이거나 분석적인 작업일 때 – 리서치, 콘텐츠 생성, 데이터 해석 등
4. 엄격한 구조보다 유연성이 중요한 경우 – 에이전트의 자율성과 융통성이 도움이 되는 경우
5. 출력 형식이 어느 정도 유동적인 경우 – 완벽히 정형화되지 않아도 되는 결과를 생성할 때

# 예시: 시장 분석을 위한 Research Crew
from crewai import Agent, Crew, Process, Task

# 전문 에이전트 생성
topics = "AI 기반 헬스케어 솔루션"
researcher = Agent(
    role="시장 리서치 전문가",
    goal=f"{topics}에 대한 종합적인 시장 데이터를 수집",
    backstory="시장 트렌드 분석과 데이터 수집에 능한 전문가"
)

analyst = Agent(
    role="시장 분석가",
    goal="수집된 데이터를 분석해 투자 기회를 도출",
    backstory="시장 데이터 해석과 인사이트 도출에 특화된 분석가"
)

# 각 역할별 작업 정의
research_task = Task(
    description=f"현재 {topics} 시장 조사 수행",
    expected_output="주요 기업, 시장 규모, 성장 동향 등이 포함된 보고서",
    agent=researcher
)

analysis_task = Task(
    description="수집된 데이터를 분석하고 상위 3개 투자 기회 도출",
    expected_output="3개 추천 투자 기회와 그 근거가 포함된 분석 보고서",
    agent=analyst,
    context=[research_task]
)

# Crew 생성 및 실행
market_analysis_crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True
)

result = market_analysis_crew.kickoff()

Flows를 선택할 때

Flows는 다음과 같은 경우에 적합합니다:

1. 정밀한 실행 제어가 필요한 경우 – 정확한 순서와 상태 관리가 중요한 워크플로우
2. 복잡한 상태 유지가 필요한 애플리케이션 – 여러 단계에 걸쳐 상태를 유지하고 변환해야 할 때
3. 예측 가능하고 구조화된 출력이 필요한 경우 – 일관된 포맷과 정확성이 요구될 때
4. 조건 분기 로직이 포함된 워크플로우 – 중간 결과에 따라 경로가 달라지는 경우
5. AI와 절차적 코드의 통합이 필요한 경우 – 전통적 로직과 AI 기능이 결합되어야 할 때

# 예시: 고객지원용 Flow
from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

# 상태 정의
class SupportTicketState(BaseModel):
    ticket_id: str = ""
    customer_name: str = ""
    issue_description: str = ""
    category: str = ""
    priority: str = "medium"
    resolution: str = ""
    satisfaction_score: int = 0

class CustomerSupportFlow(Flow[SupportTicketState]):
    @start()
    def receive_ticket(self):
        self.state.ticket_id = "TKT-12345"
        self.state.customer_name = "Alex Johnson"
        self.state.issue_description = "결제 후 프리미엄 기능을 사용할 수 없음"
        return "티켓 접수 완료"

    @listen(receive_ticket)
    def categorize_ticket(self, _):
        from crewai import LLM
        llm = LLM(model="openai/gpt-4o-mini")

        prompt = f"""
        아래 고객 이슈를 다음 카테고리 중 하나로 분류하세요:
        - Billing
        - Account Access
        - Technical Issue
        - Feature Request
        - Other

        이슈: {self.state.issue_description}

        분류 결과만 반환하세요.
        """

        self.state.category = llm.call(prompt).strip()
        return self.state.category

    @router(categorize_ticket)
    def route_by_category(self, category):
        return category.lower().replace(" ", "_")

    @listen("billing")
    def handle_billing_issue(self):
        self.state.priority = "high"
        return "결제 이슈 처리 완료"

    @listen("account_access")
    def handle_access_issue(self):
        self.state.priority = "high"
        return "계정 접근 이슈 처리 완료"

    @listen("billing", "account_access", "technical_issue", "feature_request", "other")
    def resolve_ticket(self, resolution_info):
        self.state.resolution = f"이슈 해결됨: {resolution_info}"
        return self.state.resolution

# 실행
support_flow = CustomerSupportFlow()
result = support_flow.kickoff()

Crews와 Flows를 함께 사용할 때

가장 복잡하고 정교한 애플리케이션은 종종 Crews와 Flows를 함께 활용하는 구조에서 가장 큰 이점을 얻습니다.

1. 복잡한 다단계 프로세스 – 전체 흐름은 Flow로 구성하고, 세부 작업은 Crew로 처리
2. 창의성과 구조를 동시에 요구하는 경우 – 창의적인 부분은 Crew, 구조적 처리 단계는 Flow에 맡김
3. 엔터프라이즈급 AI 애플리케이션 – 상태 관리 및 처리 흐름은 Flow, 세부 전문 작업은 Crew로 분리

# 예시: 콘텐츠 생성 파이프라인 (Crew + Flow 결합)
from crewai.flow.flow import Flow, listen, start
from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel
from typing import Dict

class ContentState(BaseModel):
    topic: str = ""
    target_audience: str = ""
    content_type: str = ""
    outline: Dict = {}
    draft_content: str = ""
    final_content: str = ""
    seo_score: int = 0

class ContentProductionFlow(Flow[ContentState]):
    @start()
    def initialize_project(self):
        self.state.topic = "지속가능한 투자"
        self.state.target_audience = "MZ 세대 투자자"
        self.state.content_type = "블로그 글"
        return "초기화 완료"

    @listen(initialize_project)
    def create_outline(self, _):
        researcher = Agent(
            role="콘텐츠 리서처",
            goal=f"{self.state.target_audience} 대상 {self.state.topic} 리서치",
            backstory="콘텐츠 기획에 최적화된 리서치 전문가"
        )

        outliner = Agent(
            role="콘텐츠 전략가",
            goal=f"{self.state.content_type}에 적합한 구조 설계",
            backstory="읽는 이를 끌어당기는 구조를 설계하는 전략가"
        )

        research_task = Task(
            description=f"{self.state.target_audience} 관심사 기반 {self.state.topic} 리서치",
            expected_output="핵심 포인트 및 통계 포함 리서치 노트",
            agent=researcher
        )

        outline_task = Task(
            description=f"{self.state.topic}에 대한 {self.state.content_type} 아웃라인 작성",
            expected_output="구간별 주요 포인트를 포함한 상세 목차",
            agent=outliner,
            context=[research_task]
        )

        outline_crew = Crew(
            agents=[researcher, outliner],
            tasks=[research_task, outline_task],
            process=Process.sequential,
            verbose=True
        )

        result = outline_crew.kickoff()

        import json
        try:
            self.state.outline = json.loads(result.raw)
        except:
            self.state.outline = {"sections": result.raw}

        return "아웃라인 생성 완료"

    @listen(create_outline)
    def write_content(self, _):
        writer = Agent(
            role="콘텐츠 작가",
            goal=f"{self.state.target_audience}을 위한 콘텐츠 작성",
            backstory="매력적인 글을 쓰는 데 능한 작가"
        )

        editor = Agent(
            role="콘텐츠 편집자",
            goal="글을 명확하고 매끄럽게 다듬기",
            backstory="콘텐츠 완성도를 높이는 데 탁월한 편집자"
        )

        writing_task = Task(
            description=f"다음 아웃라인을 바탕으로 {self.state.content_type} 작성: {self.state.outline}",
            expected_output="마크다운 형식의 초안",
            agent=writer
        )

        editing_task = Task(
            description="초안 콘텐츠를 감수 및 개선",
            expected_output="최종 마크다운 콘텐츠",
            agent=editor,
            context=[writing_task]
        )

        writing_crew = Crew(
            agents=[writer, editor],
            tasks=[writing_task, editing_task],
            process=Process.sequential,
            verbose=True
        )

        result = writing_crew.kickoff()
        self.state.final_content = result.raw

        return "콘텐츠 작성 완료"

    @listen(write_content)
    def optimize_for_seo(self, _):
        from crewai import LLM
        llm = LLM(model="openai/gpt-4o-mini")

        prompt = f"""
        아래 콘텐츠를 기반으로 SEO 최적화 분석을 해주세요. 키워드는 '{self.state.topic}'입니다.
        1~100점 사이로 점수를 주고, 개선을 위한 구체적 제안 3가지를 JSON 형식으로 출력하세요.

        콘텐츠: {self.state.final_content[:1000]}...

        형식:
        score
        """

        seo_analysis = llm.call(prompt)

        try:
            import json
            analysis = json.loads(seo_analysis)
            self.state.seo_score = analysis.get("score", 0)
            return analysis
        except:
            self.state.seo_score = 50
            return {"score": 50, "recommendations": ["SEO 분석 파싱 실패"]}

# 실행
content_flow = ContentProductionFlow()
result = content_flow.kickoff()

실용적 평가 프레임워크

당신의 사용사례에 가장 적합한 접근 방식을 결정하려면, 아래 단계별 평가 프레임워크를 활용해보세요:

1단계: 복잡도 평가

복잡도를 1~10점으로 평가합니다:

1. 단계 수 – 필요한 작업 단계의 수는?
   • 1~3단계: 낮은 복잡도 (1~3점)
   • 4~7단계: 중간 복잡도 (4~7점)
   • 8단계 이상: 높은 복잡도 (8~10점)
2. 구성 요소 간 의존성 – 서로 얼마나 얽혀 있는가?
   • 거의 없음: 낮은 복잡도 (1~3점)
   • 일부 의존성 있음: 중간 복잡도 (4~7점)
   • 고도로 복잡하게 연결됨: 높은 복잡도 (8~10점)
3. 조건 분기 로직 – 분기나 의사결정이 얼마나 많은가?
   • 단순 직선형: 낮은 복잡도 (1~3점)
   • 일부 분기 존재: 중간 복잡도 (4~7점)
   • 복잡한 결정 트리: 높은 복잡도 (8~10점)
4. 도메인 지식 필요 수준 – 얼마나 전문적인 지식이 필요한가?
   • 일반 상식 수준: 낮은 복잡도 (1~3점)
   • 일부 전문 지식 필요: 중간 복잡도 (4~7점)
   • 복수의 고급 도메인 지식 요구: 높은 복잡도 (8~10점)

→ 평균 점수를 계산하여 복잡도 수준을 파악하세요.

2단계: 정밀도 요구 수준 평가

정밀도를 1~10점으로 평가합니다:

1. 출력 구조 – 출력이 얼마나 구조화되어야 하나요?
   • 자유 텍스트: 낮은 정밀도 (1~3점)
   • 반구조화 (예: 마크다운): 중간 정밀도 (4~7점)
   • 엄격한 구조 (JSON, XML): 높은 정밀도 (8~10점)
2. 정확성 요구 – 사실 정확성이 얼마나 중요한가요?
   • 창작 콘텐츠 중심: 낮은 정밀도 (1~3점)
   • 정보 제공 중심: 중간 정밀도 (4~7점)
   • 중요한 사실 기반 정보: 높은 정밀도 (8~10점)
3. 재현성 – 반복 시 결과가 동일해야 하나요?
   • 어느 정도 달라도 무방: 낮은 정밀도 (1~3점)
   • 일부 일관성 필요: 중간 정밀도 (4~7점)
   • 완벽한 동일성 필요: 높은 정밀도 (8~10점)
4. 오류 허용도 – 오류가 미치는 영향은?
   • 낮음: 낮은 정밀도 (1~3점)
   • 보통: 중간 정밀도 (4~7점)
   • 매우 높음: 높은 정밀도 (8~10점)

→ 평균 점수를 계산하여 정밀도 수준을 파악하세요.

3단계: 매트릭스에 매핑하기

복잡도/정밀도 점수에 따라 다음과 같이 선택하세요:

• 낮은 복잡도(1~4), 낮은 정밀도(1~4): 단순 Crew
• 낮은 복잡도(1~4), 높은 정밀도(5~10): Flows (단순 LLM 호출 중심)
• 높은 복잡도(5~10), 낮은 정밀도(1~4): 복합 Crew
• 높은 복잡도(5~10), 높은 정밀도(5~10): Flows + Crew 조합

4단계: 추가 요소 고려하기

1. 개발 속도 – Crew는 프로토타입 제작이 빠름
2. 유지보수성 – Flow는 장기적 유지에 적합
3. 팀의 숙련도 – 팀이 어떤 구조에 익숙한지 고려
4. 확장성 요구 – Flow 기반 구조는 복잡한 작업에 더 잘 확장됨
5. 외부 시스템과의 연동 – 시스템 통합 요건이 있을 경우 Flow가 유리할 수 있음

결론

Crews와 Flows 중 무엇을 선택할지, 또는 둘을 어떻게 조합할지는 AI 시스템의 효과성, 유지보수성, 확장성에 직접적인 영향을 미치는 핵심 설계 결정입니다.

복잡도와 정밀도 기준으로 요구사항을 평가하면, 자신에게 맞는 구조를 체계적으로 선택할 수 있습니다.

처음에는 단순한 구조로 시작하고, 점차 애플리케이션이 성숙해질수록 아키텍처를 진화시키는 전략이 가장 좋습니다.