Crew 구성과 실행

crewAI에서 크루(Crew)는 여러 에이전트가 협력해 태스크 목록을 달성하는 단위예요. 크루 하나가 태스크 실행 전략·에이전트 협업 방식·전체 워크플로우까지 정의하니까, 크루를 어떻게 구성하느냐가 곧 오케스트레이션이 어떻게 돌아갈지를 결정합니다. 여기서는 크루 속성, 생성 방법(JSONC·파이썬 코드), 출력 처리, 체크포인팅, 실행 방식까지 차례로 살펴볼게요.

출처: 공식문서

본문

개요

crewAI에서 크루는 함께 태스크 집합을 달성하기 위해 협력하는 에이전트 그룹을 나타냅니다. 각 크루는 태스크 실행 전략, 에이전트 협업, 전체 워크플로우를 정의합니다.

크루 속성

속성	파라미터	설명
Tasks	`tasks`	크루에 할당된 태스크 목록
Agents	`agents`	크루에 속한 에이전트 목록
Process (선택)	`process`	크루가 따르는 프로세스 흐름(예: `sequential`, `hierarchical`). 기본값 `sequential`
Verbose (선택)	`verbose`	실행 중 로깅 상세 수준. 기본값 `False`
Manager LLM (선택)	`manager_llm`	계층 프로세스에서 매니저 에이전트가 쓰는 언어모델. 계층 프로세스 사용 시 필수
Function Calling LLM (선택)	`function_calling_llm`	설정하면 크루 전체 에이전트의 툴 함수 호출에 이 LLM 사용. 각 에이전트는 자체 LLM으로 재정의 가능
Config (선택)	`config`	크루 선택 설정. `Json` 또는 `Dict[str, Any]` 형식
Max RPM (선택)	`max_rpm`	실행 중 크루가 지키는 분당 최대 요청 수. 기본값 `None`
Memory (선택)	`memory`	실행 메모리 저장(단기·장기·엔티티 메모리)에 사용
Cache (선택)	`cache`	툴 실행 결과 캐시 사용 여부. 기본값 `True`
Embedder (선택)	`embedder`	크루가 사용할 임베더 설정. 대부분 메모리용. 기본값 `{"provider": "openai"}`
Step Callback (선택)	`step_callback`	각 에이전트의 매 스텝 후 호출되는 함수. 에이전트별 `step_callback`을 덮어쓰지 않음
Task Callback (선택)	`task_callback`	각 태스크 완료 후 호출되는 함수. 모니터링·후처리에 유용
Share Crew (선택)	`share_crew`	크루 전체 정보·실행을 crewAI 팀과 공유해 라이브러리 개선·모델 학습에 쓰게 할지
Output Log File (선택)	`output_log_file`	`True`면 현재 디렉터리에 logs.txt로 저장, 또는 파일 경로 지정. 파일명이 .json이면 JSON 형식, 아니면 .txt. 기본값 `None`
Manager Agent (선택)	`manager_agent`	매니저로 쓸 사용자 정의 에이전트 지정
Prompt File (선택)	`prompt_file`	크루에 쓸 프롬프트 JSON 파일 경로
Planning (선택)	`planning`	크루에 플래닝 능력 추가. 크루 반복 전에 모든 데이터를 AgentPlanner로 보내 태스크를 계획하고 이를 각 태스크 설명에 추가
Planning LLM (선택)	`planning_llm`	플래닝 과정에서 AgentPlanner가 사용하는 언어모델
Knowledge Sources (선택)	`knowledge_sources`	크루 레벨에서 사용 가능한 지식 소스(모든 에이전트가 접근)
Stream (선택)	`stream`	스트리밍 출력 활성화. `CrewStreamingOutput` 객체를 반환해 청크 단위로 순회 가능. 기본값 `False`
Chat LLM (선택)	`chat_llm`	`crewai chat` CLI 상호작용을 오케스트레이션하는 LLM. 모델명 문자열 또는 `LLM` 인스턴스 허용. 기본값 `None`
Before Kickoff Callbacks (선택)	`before_kickoff_callbacks`	크루 시작 전 실행되는 콜러블 목록. 각 콜백은 inputs dict를 받고 수정 가능. `@before_kickoff` 데코레이터와 다름. 기본값 `[]`
After Kickoff Callbacks (선택)	`after_kickoff_callbacks`	크루 종료 후 실행되는 콜러블 목록. 각 콜백은 `CrewOutput`을 받고 수정 가능. `@after_kickoff` 데코레이터와 다름. 기본값 `[]`
Tracing (선택)	`tracing`	트레이싱 제어. `True`=항상 활성, `False`=항상 비활성, `None`=환경/사용자 설정 상속. 기본값 `None`
Skills (선택)	`skills`	크루 모든 에이전트에 적용할 `Path` 객체(스킬 검색 디렉터리) 또는 사전 로드된 `Skill` 객체 목록. 기본값 `None`
Security Config (선택)	`security_config`	크루 핑거프린팅·신원 관리용 `SecurityConfig` 인스턴스. 기본값 `SecurityConfig()`
Checkpoint (선택)	`checkpoint`	자동 체크포인팅 활성화. `True`=합리적 기본값, `CheckpointConfig`=완전 제어, `False`=탈퇴, `None`=상속. 기본값 `None`

**Crew Max RPM**: `max_rpm` 속성은 크루가 분당 수행할 수 있는 최대 요청 수를 정해 레이트 리밋을 피하게 합니다. 설정하면 개별 에이전트의 `max_rpm` 설정을 덮어씁니다.

크루 생성

CrewAI에서 크루를 만드는 두 가지 일반적인 방법은 JSONC 프로젝트 구성(신규 크루 권장) 과 코드에서 직접 정의입니다.

JSONC 구성(권장)

crewai create crew <name>으로 만든 새 프로젝트는 크루 레벨 설정과 태스크에 crew.jsonc를, 에이전트마다 agents/에 파일 하나를 사용합니다.

crewai run은 crew.jsonc 또는 crew.json을 자동 감지해 참조된 에이전트를 로드하고, 누락된 플레이스홀더를 물어본 뒤 크루를 킥오프합니다.

crew.jsonc 예시:

{
  "name": "Market Research Crew",
  "agents": ["researcher", "analyst"],
  "tasks": [
    {
      "name": "research",
      "description": "Research {topic} and collect the most relevant facts.",
      "expected_output": "Structured research notes about {topic}.",
      "agent": "researcher"
    },
    {
      "name": "analysis",
      "description": "Analyze the research and write a concise report.",
      "expected_output": "A markdown report with findings and recommendations.",
      "agent": "analyst",
      "context": ["research"],
      "output_file": "output/report.md"
    }
  ],
  "process": "sequential",
  "verbose": true,
  "memory": true,
  "inputs": {
    "topic": "AI Agents"
  }
}

agents의 각 문자열은 먼저 agents/<name>.jsonc를, 다음엔 agents/<name>.json을 찾아 해석합니다.

{
  "role": "{topic} Senior Researcher",
  "goal": "Find accurate and current information about {topic}.",
  "backstory": "You are a careful researcher who cites clear evidence.",
  "llm": "openai/gpt-4o",
  "tools": ["SerperDevTool"]
}

`process`가 `"sequential"`일 때 태스크는 `tasks`에 나타난 순서대로 실행됩니다.

계층 크루는 "process": "hierarchical"로 설정하고 manager_llm 또는 manager_agent를 제공하세요. manager_agent는 최상위 agents 목록에 포함되지 않은 agents/<name>.jsonc 파일을 참조할 수 있습니다.

JSON 크루 정의는 process, verbose, memory, cache, max_rpm, planning, planning_llm, manager_llm, manager_agent, function_calling_llm, output_log_file, stream, tracing, before_kickoff_callbacks, after_kickoff_callbacks 같은 크루 레벨 필드를 지원합니다.

파이썬 콜백·사용자 정의 클래스는 {"python": "module.attribute"}를 사용합니다. 사용자 정의 툴은 "custom:<name>"으로 지정하고 런타임에 tools/<name>.py를 로드합니다.

JSON 크루 프로젝트는 신뢰하는 소스에서만 실행하세요. `custom:` 툴과 `{"python": "module.attribute"}` 참조는 크루 로드 시 로컬 파이썬 코드를 실행합니다.

클래식 Python/YAML 구성

crewai create crew <name> --classic으로 만든 클래식 프로젝트는 crew.py, config/agents.yaml, config/tasks.yaml, 그리고 @CrewBase, @agent, @task, @crew 데코레이터를 사용합니다. 이 패턴은 계속 지원되며 Using Annotations에 문서화되어 있습니다.

코드 직접 정의(대안)

YAML 설정 파일 없이 크루를 코드로 직접 정의할 수도 있습니다.

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
        )

위 코드 실행:

YourCrewName().crew().kickoff(inputs={})

이 예시에서 에이전트와 태스크는 데코레이터 없이 클래스 안에서 직접 정의되고, 에이전트·태스크 목록을 수동으로 관리합니다. 더 많은 제어를 제공하지만 큰 프로젝트에서는 유지보수가 어려울 수 있습니다.

크루 출력

크루 실행 결과는 CrewOutput 클래스에 담깁니다. 이 클래스는 raw 문자열, JSON, Pydantic 모델 등 다양한 형식으로 결과에 접근하는 구조적 방법을 제공하며, 최종 태스크 출력 결과·토큰 사용량·개별 태스크 출력을 포함합니다.

크루 출력 속성

속성	파라미터	타입	설명
Raw	`raw`	`str`	크루의 원본 출력. 기본 출력 형식
Pydantic	`pydantic`	`Optional[BaseModel]`	크루의 구조화 출력을 나타내는 Pydantic 모델 객체
JSON Dict	`json_dict`	`Optional[Dict[str, Any]]`	크루의 JSON 출력을 나타내는 딕셔너리
Tasks Output	`tasks_output`	`List[TaskOutput]`	크루 각 태스크의 출력을 나타내는 `TaskOutput` 객체 목록
Token Usage	`token_usage`	`Dict[str, Any]`	토큰 사용 요약(실행 중 LLM 성능 정보)

크루 출력 메서드·속성

메서드/속성	설명
json	출력 형식이 JSON이면 크루 출력의 JSON 문자열 표현 반환
to_dict	JSON·Pydantic 출력을 딕셔너리로 변환
str	Pydantic → JSON → raw 순으로 우선해 크루 출력 문자열 표현 반환

크루 출력 접근

크루가 실행되면 Crew 객체의 output 속성으로 결과에 접근할 수 있습니다.

# Example crew execution
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_article_task],
    verbose=True
)

crew_output = crew.kickoff()

# Accessing the crew output
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}")

크루 로그 접근

output_log_file을 True(Boolean) 또는 file_name(str)로 설정하면 크루 실행의 실시간 로그를 볼 수 있습니다. file_name.txt와 file_name.json 모두 이벤트 로깅을 지원합니다. True(Boolean)면 logs.txt로 저장됩니다. output_log_file이 False(Boolean) 또는 None이면 로그가 생성되지 않습니다.

# Save crew logs
crew = Crew(output_log_file = True)  # Logs will be saved as logs.txt
crew = Crew(output_log_file = file_name)  # Logs will be saved as file_name.txt
crew = Crew(output_log_file = file_name.txt)  # Logs will be saved as file_name.txt
crew = Crew(output_log_file = file_name.json)  # Logs will be saved as file_name.json

체크포인팅

체크포인팅은 크루가 핵심 이벤트(예: 태스크 완료) 후 상태를 자동 저장하게 해, 오래 걸리거나 중단된 실행을 완료된 태스크를 다시 실행하지 않고 정확히 그 지점부터 재개할 수 있게 합니다.

빠른 시작

checkpoint=True를 넘기면 합리적 기본값으로 체크포인팅을 켭니다(매 태스크 후 .checkpoints/에 저장):

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    checkpoint=True,  # saves to .checkpoints/ after every task
)

crew.kickoff(inputs={"topic": "AI trends"})

`CheckpointConfig`로 완전 제어

위치·트리거 이벤트·저장 백엔드·보존 정책을 세밀하게 제어하려면 CheckpointConfig를 사용하세요:

from crewai import Crew, Process
from crewai.state.checkpoint_config import CheckpointConfig

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    checkpoint=CheckpointConfig(
        location="./.checkpoints",       # directory for JSON files (default)
        on_events=["task_completed"],    # trigger after each task (default)
        max_checkpoints=5,               # keep only the 5 most recent checkpoints
    ),
)

crew.kickoff(inputs={"topic": "AI trends"})

체크포인트에서 재개

Crew.from_checkpoint()로 저장된 체크포인트 파일에서 크루를 복원한 뒤 kickoff()를 호출해 재개합니다:

# Resume from the most recent checkpoint
crew = Crew.from_checkpoint(".checkpoints/latest.json")
crew.kickoff()

체크포인트에서 복원할 때 `checkpoint_inputs`, `checkpoint_train`, `checkpoint_kickoff_event_id`는 자동으로 재구성됩니다 — 직접 설정할 필요가 없습니다.

`CheckpointConfig` 속성

속성	타입	기본값	설명
`location`	`str`	`"./.checkpoints"`	저장 위치. `JsonProvider`는 디렉터리 경로, `SqliteProvider`는 DB 파일 경로
`on_events`	`list[str]`	`["task_completed"]`	체크포인트 쓰기를 트리거하는 이벤트 유형. `["*"]`는 매 이벤트마다 체크포인트
`provider`	`JsonProvider \| SqliteProvider`	`JsonProvider()`	저장 백엔드. 기본 `JsonProvider`(일반 JSON 파일)
`max_checkpoints`	`int \| None`	`None`	유지할 최대 체크포인트 수. 쓰기 후 오래된 것부터 정리. `None`은 모두 유지

메모리 활용

크루는 메모리(단기·장기·엔티티 메모리)를 활용해 실행·학습을 개선할 수 있습니다. 실행 메모리를 저장·회상해 의사결정과 태스크 실행 전략에 도움을 줍니다.

캐시 활용

캐시로 툴 실행 결과를 저장해 동일 태스크를 다시 실행할 필요를 줄여 프로세스를 효율적으로 만듭니다.

크루 사용 메트릭

크루 실행 후 usage_metrics 속성으로 크루가 실행한 모든 태스크의 언어모델(LLM) 사용 메트릭을 볼 수 있습니다. total_tokens는 청구된 총액(prompt_tokens + completion_tokens)입니다. cached_prompt_tokens, cache_creation_tokens 같은 항목은 이미 그 합계에 포함된 부분집합을 설명하며 total_tokens 위에 더해지지 않습니다. 전체 계약은 Flows 개념 문서의 UsageMetrics field semantics 섹션을 참조하세요.

# Access the crew's usage metrics
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

크루 실행 프로세스

순차 프로세스: 태스크가 하나씩 실행되어 선형 작업 흐름 구성.
계층 프로세스: 매니저 에이전트가 크루를 조율해 태스크를 위임하고 진행 전 결과를 검증. 이 프로세스에는 manager_llm 또는 manager_agent가 필요하며 프로세스 흐름 검증에 필수입니다.

크루 킥오프

크루가 구성되면 kickoff() 메서드로 워크플로우를 시작합니다. 정의된 프로세스 흐름에 따라 실행이 시작됩니다.

# Start the crew's task execution
result = my_crew.kickoff()
print(result)

다양한 크루 킥오프 방법

kickoff(): 정의된 프로세스 흐름에 따라 실행 시작.
kickoff_for_each(): 제공된 각 입력 이벤트/항목에 대해 순차적으로 태스크 실행.

비동기 방법:

메서드	유형	설명
`akickoff()`	네이티브 async	전체 실행 체인에서 진짜 async/await
`akickoff_for_each()`	네이티브 async	목록 내 각 입력에 대해 네이티브 async 실행
`kickoff_async()`	스레드 기반	동기 실행을 `asyncio.to_thread`로 감쌈
`kickoff_for_each_async()`	스레드 기반	목록 내 각 입력에 대해 스레드 기반 async

고동시성 워크로드에서는 `akickoff()`와 `akickoff_for_each()`를 권장합니다. 태스크 실행·메모리 연산·지식 검색에 네이티브 async를 사용하기 때문입니다.

# Start the crew's task execution
result = my_crew.kickoff()
print(result)

# Example of using 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)

# Example of using native async with akickoff
inputs = {'topic': 'AI in healthcare'}
async_result = await my_crew.akickoff(inputs=inputs)
print(async_result)

# Example of using native async with akickoff_for_each
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = await my_crew.akickoff_for_each(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

# Example of using thread-based kickoff_async
inputs = {'topic': 'AI in healthcare'}
async_result = await my_crew.kickoff_async(inputs=inputs)
print(async_result)

# Example of using thread-based kickoff_for_each_async
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = await my_crew.kickoff_for_each_async(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

스트리밍 크루 실행

실시간 가시성 확보를 위해 스트리밍을 활성화하면 결과가 생성되는 대로 받을 수 있습니다:

# Enable streaming
crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True
)

# Iterate over streaming output
streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
    print(chunk.content, end="", flush=True)

# Access final result
result = streaming.result

특정 태스크에서 재실행

CLI 명령 replay로 특정 태스크에서 재실행할 수 있습니다. crewai replay -t <task_id>로 재실행할 task_id를 지정하세요. 킥오프는 최신 킥오프가 반환한 태스크 출력을 로컬에 저장해 재실행할 수 있게 합니다.

최신 킥오프 태스크 ID를 보려면:

crewai log-tasks-outputs

그 다음 특정 태스크에서 재실행하려면:

crewai replay -t <task_id>

이 명령들로 이전에 실행한 태스크의 컨텍스트를 유지한 채 최신 킥오프 태스크에서 재실행할 수 있습니다.

Crew 구성과 실행