콘텐츠로 이동

프로덕션 아키텍처 (Production Architecture)

CrewAI로 프로덕션 AI 애플리케이션을 만들 때, Flow부터 시작하는 걸 권장합니다. 개별 Crew나 Agent만 단독으로 돌리는 것도 가능하지만, 이들을 Flow로 감싸야 튼튼하고 확장 가능한 애플리케이션 구조가 잡혀요. 이 글은 Flow를 중심으로 한 CrewAI 프로덕션 아키텍처를 개념과 코드로 차근차근 설명합니다.

왜 Flow인가

Flow를 쓰면 세 가지가 해결돼요.

  1. 상태 관리(State Management): Flow는 애플리케이션의 각 단계에서 상태를 관리하는 내장 수단을 제공합니다. Crew 사이에 데이터를 넘기고, 컨텍스트를 유지하고, 사용자 입력을 처리할 때 반드시 필요한 부분이에요.
  2. 제어(Control): 반복문, 조건문, 분기 로직 같은 정밀한 실행 경로를 Flow로 정의할 수 있습니다. 엣지 케이스를 다루고 애플리케이션이 예측 가능하게 동작하도록 하는 데 핵심이에요.
  3. 관측성(Observability): Flow는 실행을 추적하고 문제를 디버깅하며 성능을 모니터링하기 쉬운 명확한 구조를 만들어 줍니다. 자세한 인사이트가 필요하다면 CrewAI Tracing을 추천해요. crewai login만 실행하면 무료 관측 기능이 켜집니다.

아키텍처 살펴보기

전형적인 프로덕션 CrewAI 애플리케이션은 이런 모습입니다.

1. Flow 클래스

Flow 클래스가 진입점 역할을 해요. 상태 스키마와 로직을 실행할 메서드를 정의합니다.

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

class AppState(BaseModel):
    user_input: str = ""
    research_results: str = ""
    final_report: str = ""

class ProductionFlow(Flow[AppState]):
    @start()
    def gather_input(self):
        # ... logic to get input ...
        pass

    @listen(gather_input)
    def run_research_crew(self):
        # ... trigger a Crew ...
        pass

@start로 흐름의 시작점을, @listen으로 이전 단계에 이어지는 다음 단계를 표시해요. 그러면 단계 사이의 실행 순서가 Flow 안에서 자연스럽게 연결됩니다.

2. 상태 관리

상태는 Pydantic 모델로 정의하세요. 타입 안전성을 보장하고, 각 단계에서 어떤 데이터를 쓸 수 있는지 명확하게 보여줍니다.

  • 최소한으로 유지하기: 단계 사이에 꼭 보존해야 할 것만 저장해요.
  • 구조화된 데이터 쓰기: 가능하면 비구조화된 딕셔너리는 피합니다.

3. 작업 단위로서의 Crew

복잡한 작업은 Crew에 위임하세요. Crew는 "토픽을 조사한다", "블로그 포스트를 작성한다"처럼 특정 목표 하나에 집중해야 합니다.

  • 과도하게 설계하지 않기: Crew를 한 가지 역할에 집중시키세요.
  • 상태를 명시적으로 넘기기: Flow 상태에서 필요한 데이터를 Crew 입력으로 명시적으로 전달합니다.
@listen(gather_input)
def run_research_crew(self):
    crew = ResearchCrew()
    result = crew.kickoff(inputs={"topic": self.state.user_input})
    self.state.research_results = result.raw

self.state.user_input을 Crew의 inputs로 넘기고, 그 결과의 raw를 다시 상태에 저장하는 흐름이에요. 이렇게 하면 각 단계가 어떤 입력을 받아 어떤 출력을 내놓는지가 명확해집니다.

제어 프리미티브

CrewAI의 제어 프리미티브를 활용하면 Crew에 견고함과 제어력을 더할 수 있어요.

1. 태스크 가드레일 (Task Guardrails)

Task Guardrails로 태스크 출력을 받아들이기 전에 검증합니다. 에이전트가 품질 좋은 결과를 내도록 보장하는 장치예요.

def validate_content(result: TaskOutput) -> Tuple[bool, Any]:
    if len(result.raw) < 100:
        return (False, "Content is too short. Please expand.")
    return (True, result.raw)

task = Task(..., guardrail=validate_content)

가드레일 함수는 (통과 여부, 값) 튜플을 반환합니다. 거짓이면 해당 출력은 거부되고, 참이면 그 값을 태스크 결과로 사용해요.

2. 구조화된 출력 (Structured Outputs)

태스크 사이에 데이터를 주고받을 때는 항상 구조화된 출력(output_pydantic 또는 output_json)을 쓰세요. 파싱 오류를 막고 타입 안전성을 보장합니다.

class ResearchResult(BaseModel):
    summary: str
    sources: List[str]

task = Task(..., output_pydantic=ResearchResult)

3. LLM 훅 (LLM Hooks)

LLM Hooks로 LLM에 메시지를 보내기 전에 메시지를 검사·수정하거나, 응답을 정리할 수 있어요.

@before_llm_call
def log_request(context):
    print(f"Agent {context.agent.role} is calling the LLM...")

배포 패턴

Flow를 배포할 때는 다음을 고려하세요.

CrewAI Enterprise

가장 쉬운 배포 방법은 CrewAI Enterprise를 쓰는 거예요. 인프라, 인증, 모니터링을 알아서 처리해 줍니다. 시작하려면 Deployment Guide를 확인하세요.

crewai deploy create

비동기 실행 (Async Execution)

오래 걸리는 작업은 kickoff_async를 사용해 API가 블로킹되지 않게 해요.

영속화 (Persistence)

@persist 데코레이터로 Flow 상태를 데이터베이스에 저장합니다. 프로세스가 크래시나거나 사용자 입력을 기다려야 할 때 실행을 재개할 수 있어요.

@persist
class ProductionFlow(Flow[AppState]):
    # ...

기본적으로 @persist는 kickoff(inputs={"id": <uuid>})가 주어지면 Flow를 재개하면서 같은 flow_uuid 기록을 이어갑니다. 이전 실행에서 상태를 hydrate하되 새로운 state.id 아래에 써서 새 계보(lineage)로 포크(fork) 하려면 restore_from_state_id를 넘기세요.

flow.kickoff(restore_from_state_id="<previous-run-state-id>")

새 실행은 새 state.id를 받아요(자동 생성되거나 inputs["id"]로 고정됨). 그래서 그 @persist 쓰기가 원본의 기록을 이어가지 않습니다. 이걸 from_checkpoint와 함께 쓰면 ValueError가 나요 — 비동기 하이드레이션 소스는 하나만 골라야 합니다.

정리

  • Flow부터 시작하세요.
  • 명확한 State를 정의하세요.
  • 복잡한 작업에는 Crew를 쓰세요.
  • API와 영속화와 함께 배포하세요.

실무 관점 한마디

개별 Crew를 먼저 만들고 나중에 Flow로 묶어도 되나 싶지만, 실제로는 Flow 골격을 먼저 잡고 그 안에 Crew를 채워 넣는 편이 훨씬 수월해요. 상태 스키마를 처음부터 정의해 두면 나중에 단계를 추가하거나 순서를 바꿀 때 흐름이 흐트러지지 않고, @start/@listen로 도식화된 실행 경로 덕분에 디버깅 지점도 명확해집니다. 작게 시작해서 구조를 먼저 규정하세요.

더 알아보기