CrewAI의 크루(Crew): 구조와 속성 정리

크루(Crew)란?

CrewAI에서의 크루는 여러 에이전트(Agent)들이 협력하여 하나 이상의 작업(Task)을 수행하는 협업 단위입니다. 크루는 작업 수행 전략, 에이전트 간 협업 방식, 전체 워크플로우를 정의합니다.

크루 속성

속성명	파라미터	설명
작업 목록	tasks	크루에 할당된 작업(Task)들의 리스트입니다.
에이전트 목록	agents	크루에 소속된 에이전트들의 리스트입니다.
프로세스 (선택 사항)	process	크루가 따르는 작업 수행 흐름입니다. 기본값은 sequential입니다. 예: 순차적, 계층적 등
로그 레벨 (선택 사항)	verbose	실행 중 로깅의 상세 정도를 조절합니다. 기본값은 False입니다.
매니저 LLM (선택 사항)	manager_llm	계층적 프로세스를 사용할 경우, 매니저 에이전트가 사용하는 언어 모델입니다. 계층적 프로세스에서는 필수입니다.
함수 호출용 LLM (선택 사항)	function_calling_llm	크루 내 모든 에이전트가 도구 호출에 사용할 LLM입니다. 개별 에이전트에서 별도 설정 시 해당 값이 우선 적용됩니다.
추가 설정 (선택 사항)	config	Json 또는 딕셔너리 형식으로 전달되는 추가 구성 정보입니다.
최대 요청 속도 (선택 사항)	max_rpm	분당 최대 요청 수를 제한하여 Rate Limit을 방지합니다. 지정하면 개별 에이전트의 max_rpm 설정을 덮어씁니다.
메모리 (선택 사항)	memory	실행 메모리(단기, 장기, 엔티티 메모리 등)를 저장하는 기능입니다.
메모리 설정 (선택 사항)	memory_config	사용할 메모리 공급자에 대한 설정 정보입니다.
캐시 사용 여부 (선택 사항)	cache	도구 실행 결과를 캐시에 저장하여 동일 작업의 반복 실행을 방지할 수 있습니다. 기본값은 True입니다.
임베더 설정 (선택 사항)	embedder	현재는 주로 메모리에서 사용되며, 임베딩 제공자 설정입니다. 기본값은 {"provider": "openai"}입니다.
스텝 콜백 (선택 사항)	step_callback	각 스텝 이후 호출되는 함수로, 에이전트의 행동 로깅이나 추가 작업 수행 시 활용됩니다. 에이전트 별 콜백을 덮어쓰지 않습니다.
태스크 콜백 (선택 사항)	task_callback	각 태스크 완료 시 호출되는 함수로, 모니터링이나 후처리에 사용됩니다.
크루 공유 여부 (선택 사항)	share_crew	CrewAI 팀과 크루 실행 정보를 공유하여 라이브러리 개선 및 학습에 기여할지 여부입니다.
로그 파일 출력 (선택 사항)	output_log_file	로그를 logs.txt 또는 .json 형식으로 저장합니다. True로 설정 시 현재 디렉토리에 저장됩니다.
매니저 에이전트 (선택 사항)	manager_agent	커스텀 매니저 에이전트를 설정할 수 있습니다.
프롬프트 파일 (선택 사항)	prompt_file	크루에서 사용할 프롬프트 JSON 파일의 경로입니다.
계획 기능 활성화 (선택 사항)	planning	활성화 시 AgentPlanner가 크루 데이터를 기반으로 작업 계획을 세우고 이를 태스크 설명에 추가합니다.
계획용 LLM (선택 사항)	planning_llm	AgentPlanner가 계획을 수립할 때 사용하는 언어 모델입니다.

Crew Max RPM max_rpm 속성은 크루가 분당 수행할 수 있는 최대 요청 수를 설정합니다. 설정 시 개별 에이전트의 max_rpm보다 우선합니다.

크루 생성 방식

CrewAI에서는 크루를 생성하는 두 가지 주요 방식이 있습니다:

1. YAML 설정을 통한 생성 (권장)
2. 코드 직접 정의 방식

YAML 설정 (권장)

YAML 파일 기반 구성은 크루, 에이전트, 태스크 정의를 더 깔끔하고 유지보수하기 쉽게 만들며, CrewAI의 프로젝트 구조와도 일관됩니다.

CrewBase 클래스를 상속받고, 데코레이터를 활용해 에이전트, 태스크, 크루를 정의합니다.

예시: 데코레이터 기반 크루 클래스

from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List

@CrewBase
class YourCrewName:
    """이 크루에 대한 설명"""

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

    # YAML 설정 파일 경로
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @before_kickoff
    def prepare_inputs(self, inputs):
        inputs['additional_data'] = "Some extra information"
        return inputs

    @after_kickoff
    def process_output(self, output):
        output.raw += "\nProcessed after kickoff."
        return output

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

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

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

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

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

태스크는 정의된 순서대로 실행됩니다.

CrewBase 클래스와 데코레이터를 활용하면, 에이전트와 태스크를 자동으로 수집하여 수작업을 줄이고 구조적인 관리를 할 수 있습니다.

annotations.py의 데코레이터 개요

CrewAI는 annotations.py 파일에서 여러 데코레이터를 제공하여, crew 클래스 내 메서드를 특별한 용도로 표시할 수 있습니다.

• @CrewBase: 해당 클래스를 crew의 기반 클래스임을 표시합니다.
• @agent: Agent 객체를 반환하는 메서드임을 나타냅니다.
• @task: Task 객체를 반환하는 메서드임을 나타냅니다.
• @crew: Crew 객체를 반환하는 메서드임을 나타냅니다.
• @before_kickoff (선택): crew 시작 전에 실행할 메서드를 지정합니다.
• @after_kickoff (선택): crew 종료 후 실행할 메서드를 지정합니다.

이러한 데코레이터는 에이전트와 태스크를 수동으로 리스트화하지 않고도 구조를 자동으로 수집할 수 있게 도와줍니다.

직접 코드 정의 방식 (대안)

YAML 설정 파일을 사용하지 않고, 코드 내에서 직접 crew를 정의할 수도 있습니다.

from crewai import Agent, Crew, Task, Process
from crewai_tools import YourCustomTool

class YourCrewName:
    def agent_one(self) -> Agent:
        return Agent(
            role="데이터 분석가",  # Data Analyst
            goal="시장 내 데이터 트렌드를 분석한다",  # Analyze data trends in the market
            backstory="경제학 배경을 가진 경험 많은 데이터 분석가",  # An experienced data analyst with a background in economics
            verbose=True,
            tools=[YourCustomTool()]
        )

    def agent_two(self) -> Agent:
        return Agent(
            role="시장 조사원",  # Market Researcher
            goal="시장 동향에 대한 정보를 수집한다",  # Gather information on market dynamics
            backstory="세밀한 관찰력을 가진 성실한 리서처",  # A diligent researcher with a keen eye for detail
            verbose=True
        )

    def task_one(self) -> Task:
        return Task(
            description="최근 시장 데이터를 수집하고 트렌드를 식별한다.",  # Collect recent market data and identify trends.
            expected_output="시장 주요 트렌드를 요약한 보고서",  # A report summarizing key trends in the market.
            agent=self.agent_one()
        )

    def task_two(self) -> Task:
        return Task(
            description="시장 동향에 영향을 미치는 요인을 조사한다.",  # Research factors affecting market dynamics.
            expected_output="시장에 영향을 주는 요인에 대한 분석",  # An analysis of factors influencing the market.
            agent=self.agent_two()
        )

    def crew(self) -> Crew:
        return Crew(
            agents=[self.agent_one(), self.agent_two()],
            tasks=[self.task_one(), self.task_two()],
            process=Process.sequential,
            verbose=True
        )

• 위 예제에서는 데코레이터 없이 클래스 내에서 직접 agent와 task를 정의하고, 수동으로 리스트화하여 crew에 전달합니다.
• 이 방식은 더 많은 제어권을 제공하지만, 프로젝트 규모가 커질수록 유지보수가 어려워질 수 있습니다.

Crew Output 구조

CrewAI의 실행 결과는 CrewOutput 클래스에 캡슐화되어 제공됩니다. 이 클래스는 raw 문자열, JSON, Pydantic 모델 등의 다양한 형식으로 결과를 구조화하여 제공합니다.

CrewOutput 주요 속성

속성명	파라미터	타입	설명
Raw	raw	str	crew의 실행 결과를 문자열 형태로 제공. 기본 출력 형식.
Pydantic	pydantic	Optional[BaseModel]	결과를 구조화한 Pydantic 모델.
JSON Dict	json_dict	Optional[Dict[str, Any]]	JSON 딕셔너리 형태의 결과.
Tasks Output	tasks_output	List[TaskOutput]	각 task의 개별 실행 결과 리스트.
Token Usage	token_usage	Dict[str, Any]	토큰 사용량 정보를 포함. 모델 성능 분석에 활용 가능.

CrewOutput 메서드 및 속성

메서드/속성	설명
json	결과를 JSON 문자열로 반환.
to_dict	JSON 및 Pydantic 결과를 딕셔너리로 변환.
__str__()	Pydantic → JSON → raw 순서로 문자열 출력값을 반환.

Crew Output 접근 방법

Crew가 실행되면 Crew 객체의 output 속성을 통해 결과를 확인할 수 있습니다. 이 결과는 CrewOutput 클래스를 통해 다양한 방식으로 접근할 수 있습니다.

예제

# Crew 실행 예시
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_article_task],
    verbose=True
)

crew_output = crew.kickoff()

# 출력 접근
print(f"Raw Output: {crew_output.raw}")
if crew_output.json_dict:
    print(f"JSON Output: {json.dumps(crew_output.json_dict, indent=2)}")
if crew_output.pydantic:
    print(f"Pydantic Output: {crew_output.pydantic}")
print(f"Tasks Output: {crew_output.tasks_output}")
print(f"Token Usage: {crew_output.token_usage}")

Crew 로그 저장

output_log_file 속성을 True 또는 파일명(str)으로 설정하면 실행 로그를 실시간으로 저장할 수 있습니다.

• True 설정 시 logs.txt로 저장됩니다.
• 파일명을 지정하면 .txt 또는 .json으로 저장됩니다.
• False 또는 None일 경우 로그는 저장되지 않습니다.

# 로그 저장 예시
crew = Crew(output_log_file=True)  # logs.txt로 저장됨
crew = Crew(output_log_file="crew_log")  # crew_log.txt로 저장됨
crew = Crew(output_log_file="crew_log.json")  # JSON 형식으로 저장됨

메모리 활용

Crew는 단기, 장기, 엔터티 메모리를 활용하여 학습 및 실행 능력을 향상시킬 수 있습니다. 이를 통해 과거 실행 기록을 기억하고, 더 나은 의사결정 및 작업 전략 수립이 가능합니다.

캐시 활용

도구 실행 결과를 캐시에 저장하면 동일한 작업을 다시 실행할 필요 없이 빠르게 결과를 재사용할 수 있어 성능 향상에 기여합니다.

Crew 사용량 측정

usage_metrics 속성을 통해 Crew가 실행한 각 Task에 대한 LLM 사용량을 확인할 수 있습니다. 이는 효율성 분석 및 최적화에 유용합니다.

# 사용량 측정 예시
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

Crew 실행 방식

• 순차적 실행 (Sequential): Task가 하나씩 순서대로 실행됨.
• 계층적 실행 (Hierarchical): 매니저 역할의 agent가 작업을 조정하고 결과를 검토 후 다음 단계 진행.
  • 이 방식은 manager_llm 또는 manager_agent 설정이 필요함.

Crew 시작 (kickoff())

crew를 구성한 후 kickoff() 메서드로 실행을 시작할 수 있습니다.

# 기본 kickoff
result = my_crew.kickoff()
print(result)

다양한 kickoff 방식

• kickoff(): 정의된 흐름대로 실행
• kickoff_for_each(): 입력 배열에 대해 순차 실행
• kickoff_async(): 비동기로 실행
• kickoff_for_each_async(): 입력 배열에 대해 비동기 병렬 실행

# 다양한 kickoff 예시
result = my_crew.kickoff()
print(result)

# kickoff_for_each 예시
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
results = my_crew.kickoff_for_each(inputs=inputs_array)
for result in results:
    print(result)

# kickoff_async 예시
inputs = {'topic': 'AI in healthcare'}
async_result = my_crew.kickoff_async(inputs=inputs)
print(async_result)

# kickoff_for_each_async 예시
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = my_crew.kickoff_for_each_async(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

특정 Task에서 재실행 (Replay)

CrewAI는 CLI 명령어 crewai replay -t <task_id>를 통해 특정 task부터 재실행할 수 있는 기능을 제공합니다.

kickoff() 시점에 저장된 최신 task 출력이 로컬에 기록되어 재실행에 활용됩니다.

CLI를 사용한 특정 Task 재실행 방법

다음 단계를 따라 재실행 기능을 사용할 수 있습니다:

1. 터미널이나 명령 프롬프트를 엽니다.
2. CrewAI 프로젝트가 위치한 디렉토리로 이동합니다.
3. 다음 명령어를 실행하여 최근 kickoff된 task ID들을 확인합니다:

crewai log-tasks-outputs

1. 특정 task에서 재실행하려면 다음 명령어를 사용합니다:

crewai replay -t <task_id>

이 명령어들을 통해 최근 kickoff된 task의 실행 맥락을 유지하면서 특정 task부터 다시 실행할 수 있습니다.