CrewAI에서 일(Task) 다루기 (Tasks)¶

에이전트가 "무엇을 해야 하는지"를 정해 주는 게 Task예요. CrewAI에서 Task는 하나의 구체적인 업무 단위이고, 실제로 그 일을 수행하는 건 Agent죠. 역할을 다시 정리하면 이렇게 돼요. 범죄 수사 팀을 상상해 볼게요. 탐정(Agent)이 현장을 조사하는 사람이라면, Task는 "이 사건의 용의자를 조사해 와"라는 지시서예요. 지시서에는 뭘 조사할지, 그 결과물이 어떤 모양이어야 하는지, 누가 할지가 적혀 있어야 하죠. CrewAI의 Task도 마찬가지예요. 설명(description), 담당 에이전트(agent), 필요한 도구(tools) 같은 실행에 필요한 정보를 담고 있어요.

Task는 반드시 혼자 해야 하는 건 아니에요. 여러 에이전트가 함께 협업해서 하나의 Task를 완성하도록 만들 수도 있어요. 이건 Task가 가진 속성과 Crew의 실행 프로세스(process)가 조합되면서 이루어지는데, 팀의 협업과 효율을 높이는 게 목적이에요.

작업 실행 흐름 (Task Execution Flow)¶

Task가 어떤 순서로 실행될지는 Crew를 만들 때 정해요. 실행 방식은 크게 두 가지예요.

순차(Sequential): 정의한 순서 그대로 하나씩 실행해요.
계층(Hierarchical): 각 에이전트의 역할과 전문성에 맞춰서 Task를 배정해요.

Crew를 만들 때 process 인자로 어느 쪽인지 정해요.

from crewai import Crew, Process

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential  # 또는 Process.hierarchical
)

Task가 가진 속성 (Task Attributes)¶

Task 객체에 넣을 수 있는 속성들을 정리하면 이래요. 꼭 필요한 건 description과 expected_output 두 가지예요.

속성	파라미터	타입	설명
Description	`description`	`str`	Task가 무엇을 하는지 명확하고 간결하게 적은 문장이에요.
Expected Output	`expected_output`	`str`	Task가 끝났을 때 결과물이 어떤 모습인지 상세히 적어요.
Name (선택)	`name`	`Optional[str]`	Task를 식별하기 위한 이름이에요.
Agent (선택)	`agent`	`Optional[BaseAgent]`	이 Task를 실행할 에이전트예요.
Tools (선택)	`tools`	`List[BaseTool]`	이 Task에서만 사용할 수 있도록 제한한 도구 목록이에요.
Context (선택)	`context`	`Optional[List["Task"]]`	이 Task의 맥락으로 사용할 다른 Task들의 출력물이에요.
Async Execution (선택)	`async_execution`	`Optional[bool]`	비동기로 실행할지 여부예요. 기본값은 `False`예요.
Human Input (선택)	`human_input`	`Optional[bool]`	에이전트의 최종 답을 사람이 검토할지 여부예요. 기본값은 `False`예요.
Markdown (선택)	`markdown`	`Optional[bool]`	최종 답을 Markdown 형식으로 내도록 지시할지예요. 기본값은 `False`예요.
Config (선택)	`config`	`Optional[Dict[str, Any]]`	Task에 특화된 설정 파라미터예요.
Output File (선택)	`output_file`	`Optional[str]`	Task 결과를 저장할 파일 경로예요.
Create Directory (선택)	`create_directory`	`Optional[bool]`	`output_file`의 디렉터리가 없으면 만들지 여부예요. 기본값은 `True`예요.
Output JSON (선택)	`output_json`	`Optional[Type[BaseModel]]`	JSON 출력 구조를 잡아 줄 Pydantic 모델이에요.
Output Pydantic (선택)	`output_pydantic`	`Optional[Type[BaseModel]]`	Task 출력용 Pydantic 모델이에요.
Callback (선택)	`callback`	`Optional[Any]`	Task 완료 후 실행할 함수나 객체예요.
Guardrail (선택)	`guardrail`	`Optional[Callable]`	다음 Task로 넘어가기 전에 출력을 검증하는 함수예요.
Guardrails (선택)	`guardrails`	`Optional[List[Callable]]`	출력을 검증할 가드레일 목록이에요.
Guardrail Max Retries (선택)	`guardrail_max_retries`	`Optional[int]`	가드레일 검증이 실패했을 때 재시도할 최대 횟수예요. 기본값은 `3`이에요.

Task 만들기¶

CrewAI에서 Task를 만드는 방법은 크게 두 갈래예요. 새 크루를 만들 때 권장되는 JSONC 프로젝트 설정과, 코드에 직접 정의하는 방법이에요. 예전 방식인 YAML 설정도 계속 지원돼요.

JSONC 설정 (권장)¶

crewai create crew <name>으로 만든 새 프로젝트는 작업 정의를 crew.jsonc에 둬요. agents 배열은 agents/ 폴더 안의 파일을 가리키고, tasks 배열은 크루가 실행할 순서 있는 작업들을 정의해요. 두 개의 순서 있는 Task가 든 crew.jsonc 예시를 볼게요.

{
  "name": "Research Crew",
  "agents": ["researcher", "reporting_analyst"],
  "tasks": [
    {
      "name": "research_task",
      "description": "Conduct thorough research about {topic}. Include current and relevant information.",
      "expected_output": "A list of the most relevant information about {topic}.",
      "agent": "researcher"
    },
    {
      "name": "reporting_task",
      "description": "Review the research and expand it into a detailed report.",
      "expected_output": "A polished markdown report without fenced code blocks.",
      "agent": "reporting_analyst",
      "context": ["research_task"],
      "markdown": true,
      "output_file": "report.md"
    }
  ],
  "inputs": {
    "topic": "AI Agents"
  }
}

각 Task는 description과 expected_output을 반드시 포함해야 해요. agent 값은 agents에 등록된 에이전트 이름과 일치해야 하고, context는 앞선 Task 이름들의 목록이에요. 여기서 주의할 점은, 미래의 Task를 앞당겨 참조하는 것(forward reference)은 거부돼요. 순차적인 맥락이 명확하게 유지되도록 하려는 의도예요. Task 항목은 공개된 Task 필드를 대부분 지원하는데, 흔히 쓰는 것만 추리면 name, agent, context, output_file, tools, human_input, async_execution, guardrail, guardrails, guardrail_max_retries, markdown, input_files, output_json, output_pydantic, response_model, converter_cls예요. 조건부로 실행할 Task가 필요하면 "type": "ConditionalTask"와 condition 필드를 쓰면 돼요.

코드에 직접 정의 (대안)¶

YAML 설정 없이 코드에서 Task를 바로 정의할 수도 있어요.

from crewai import Task

research_task = Task(
    description="""
    Conduct a thorough research about AI Agents.
    Make sure you find any interesting and relevant information
    given the current year is 2025.
    """,
    expected_output="""
    A list with 10 bullet points of the most relevant information about AI Agents
    """,
    agent=researcher
)

reporting_task = Task(
    description="""
    Review the context you got and expand each topic into a full section
    for a report. Make sure the report is detailed and contains any
    and all relevant information.
    """,
    expected_output="""
    A fully fledge reports with the mains topics, each with a full section of information.
    """,
    agent=reporting_analyst,
    markdown=True,          # 최종 출력에 마크다운 서식 적용
    output_file="report.md"
)

Task 출력 (Task Output)¶

에이전트가 생성한 결과물을 다루는 건 AI 워크플로를 만들 때 핵심이에요. CrewAI는 결과를 TaskOutput 클래스 안에 담아 줘서, 구조적으로 접근할 수 있게 해요. 이 덕분에 Task 사이에서 결과를 주고받기도 쉬워요.

기본적으로 TaskOutput에는 raw(원본 출력)만 들어 있어요. 그리고 pydantic이나 json_dict 같은 구조화된 출력은, 원래 Task 객체가 output_pydantic이나 output_json으로 설정되어 있었을 때만 포함돼요. 처음엔 그냥 "결과물이 하나의 객체로 감싸져서 나온다" 정도로 이해하면 충분해요.

주요 속성을 보면 이런 것들이 있어요.

속성	타입	설명
Description	`str`	Task의 설명이에요.
Summary	`Optional[str]`	설명의 첫 10개 단어로 자동 생성된 요약이에요.
Raw	`str`	Task의 원본 출력이에요. 기본 출력 형식이에요.
Pydantic	`Optional[BaseModel]`	구조화된 출력을 나타내는 Pydantic 모델 객체예요.
JSON Dict	`Optional[Dict[str, Any]]`	JSON 출력을 나타내는 딕셔너리예요.
Agent	`str`	Task를 실행한 에이전트예요.
Output Format	`OutputFormat`	RAW, JSON, Pydantic 중 출력 형식이에요. 기본값은 RAW예요.
Messages	`list[LLMMessage]`	마지막 실행에서 나온 메시지들이에요.

여기에 붙어 있는 유용한 메서드도 있어요.

json: 출력 형식이 JSON일 때, Task 출력의 JSON 문자열을 돌려줘요.
to_dict: JSON과 Pydantic 출력을 딕셔너리로 변환해요.
str: Pydantic, JSON, raw 순으로 우선순위를 두고 문자열 표현을 돌려줘요.

Task가 실행되고 나면 Task 객체의 output 속성으로 출력에 접근할 수 있어요.

# 예시 Task
task = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool]
)

# 크루 실행
crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)
result = crew.kickoff()

# Task 출력 접근
task_output = task.output
print(f"Task Description: {task_output.description}")
print(f"Task Summary: {task_output.summary}")
print(f"Raw Output: {task_output.raw}")

if task_output.json_dict:
    print(f"JSON Output: {json.dumps(task_output.json_dict, indent=2)}")

if task_output.pydantic:
    print(f"Pydantic Output: {task_output.pydantic}")

Markdown 출력 포맷 (Markdown Output Formatting)¶

에이전트가 최종 답을 Markdown 문법으로 다듬어 내게 하려면 markdown 파라미터를 True로 켜요. 이렇게 하면 Task가 에이전트에게 "최종 답은 정해진 Markdown 규칙으로 포맷해"라고 별도의 지시를 더해요.

formatted_task = Task(
    description="Create a comprehensive report on AI trends",
    expected_output="A well-structured report with headers, sections, and bullet points",
    agent=reporter_agent,
    markdown=True  # 자동 마크다운 포맷 활성화
)

markdown=True로 켜면 에이전트에게 이런 규칙들이 추가로 전달돼요.

# : 제목(헤더)
**text** : 굵은 글씨
*text* : 기울임 글씨
- 또는 * : 불릿 포인트
`code` : 인라인 코드
```language ``` : 코드 블록

이 설정의 장점을 꼽자면, 모든 출력이 일관된 마크다운 규칙을 따르고, 구조화된 콘텐츠라 읽기 좋고, 문서화 시스템에 바로 넣을 수 있고, 마크다운은 어디서나 지원되니 플랫폼 간 호환도 좋아요.

Task 의존성과 맥락 (Task Dependencies and Context)¶

Task가 다른 Task의 출력 결과에 의존하게 만들고 싶다면 context 속성을 써요. context에 넣은 Task는, 그 Task가 끝날 때까지 기다렸다가 그 출력을 맥락으로 사용해요.

research_task = Task(
    description="Research the latest developments in AI",
    expected_output="A list of recent AI developments",
    agent=researcher
)

analysis_task = Task(
    description="Analyze the research findings and identify key trends",
    expected_output="Analysis report of AI trends",
    agent=analyst,
    context=[research_task]  # 이 Task는 research_task가 끝나기를 기다려요
)

Task 가드레일 (Task Guardrails)¶

가드레일은 Task의 산출물이 다음 Task로 넘어가기 전에 검증하고 필요하면 변형하는 장치예요. 출력 품질을 보장하고, 기준에 미달하면 에이전트에게 피드백을 줘서 고치게 하는 역할을 해요. CrewAI는 두 종류의 가드레일을 지원해요.

함수 기반 가드레일: 검증 로직을 직접 짠 Python 함수예요. 검증 과정을 완전히 제어할 수 있고, 결과가 확정적(deterministic)이라 믿을 수 있어요.
LLM 기반 가드레일: 검증 기준을 자연어 문자열로 적어요. 에이전트의 LLM이 그 기준에 맞는지 판단해요. 복잡하거나 주관적인 검증에 잘 맞아요.

함수 기반 가드레일¶

가드레일 함수는 guardrail 파라미터에 넘겨요.

from typing import Tuple, Any
from crewai import TaskOutput

def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    """블로그 콘텐츠가 요구사항을 충족하는지 검증한다."""
    try:
        # 단어 수 확인
        word_count = len(result.raw.split())
        if word_count > 200:
            return (False, "Blog content exceeds 200 words")
        # 추가 검증 로직은 여기에
        return (True, result.raw.strip())
    except Exception as e:
        return (False, "Unexpected error during validation")

blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail=validate_blog_content  # 가드레일 함수 추가
)

가드레일 함수의 요구사항을 정리하면 이래요.

함수 시그니처: 인자를 정확히 하나(태스크 출력) 받아야 하고, (bool, Any) 튜플을 돌려줘야 해요. 타입 힌트는 권장이지만 필수는 아니에요.
반환 값: 성공 시 (True, validated_result)처럼 (bool, Any)를, 실패 시 (False, "Error message...")처럼 (bool, str)을 돌려줘요.

LLM 기반 가드레일 (문자열 설명)¶

검증 함수를 직접 짜는 대신, 자연어 기준을 문자열로 주는 방법도 있어요. guardrail이나 guardrails에 문자열을 넘기면 CrewAI가 자동으로 LLMGuardrail을 만들어서, 에이전트의 LLM이 그 기준대로 출력을 검증해요.

이때 두 가지 전제 조건이 있어요.

Task에 agent가 지정되어 있어야 해요. (가드레일이 에이전트의 LLM을 쓰거든요.)
검증 기준을 명확하게 설명하는 문자열이 필요해요.

from crewai import Task

# 단일 LLM 기반 가드레일
blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail="The blog post must be under 200 words and contain no technical jargon"
)

LLM 기반 가드레일은 이런 경우에 특히 유용해요.

프로그램으로 표현하기 어려운 복잡한 검증 로직
어조(tone), 스타일, 품질 평가 같은 주관적 기준
코드로 쓰기보다 자연어로 표현하는 게 쉬운 요구사항

LLM 가드레일의 동작은 이렇게 진행돼요.

Task 출력을 우리가 준 기준과 대조해서 분석해요.
기준에 부합하면 (True, output)을 반환해요.
실패하면 구체적인 피드백과 함께 (False, feedback)을 반환해요.

여러 가드레일 (Multiple Guardrails)¶

guardrails 파라미터로 가드레일을 여럿 적용할 수 있어요. 이때 각 가드레일은 순차적으로 실행되면서, 앞선 가드레일이 통과시킨 출력을 다음 가드레일이 받아요. 덕분에 검증과 변형 단계를 사슬처럼 연결할 수 있어요.

guardrails에 넣을 수 있는 건 다음과 같아요.

가드레일 함수나 문자열 설명의 리스트
단일 가드레일 함수나 문자열 (guardrail과 동일한 역할)

한 가지 주의할 점은, guardrails가 설정되면 guardrail보다 우선한다는 거예요. 즉 guardrails가 있으면 guardrail 파라미터는 무시돼요.

from typing import Tuple, Any
from crewai import TaskOutput, Task

def validate_word_count(result: TaskOutput) -> Tuple[bool, Any]:
    """단어 수가 범위 안인지 검증한다."""
    word_count = len(result.raw.split())
    if word_count < 100:
        return (False, f"Content too short: {word_count} words. Need at least 100 words.")
    if word_count > 500:
        return (False, f"Content too long: {word_count} words. Maximum is 500 words.")
    return (True, result.raw)

def validate_no_profanity(result: TaskOutput) -> Tuple[bool, Any]:
    """부적절한 언어가 있는지 확인한다."""
    profanity_words = ["badword1", "badword2"]  # 예시 목록
    content_lower = result.raw.lower()
    for word in profanity_words:
        if word in content_lower:
            return (False, f"Inappropriate language detected: {word}")
    return (True, result.raw)

def format_output(result: TaskOutput) -> Tuple[bool, Any]:
    """출력을 포맷하고 정리한다."""
    formatted = result.raw.strip()
    # 첫 글자를 대문자로
    formatted = formatted[0].upper() + formatted[1:] if formatted else formatted
    return (True, formatted)

# 여러 가드레일을 순차적으로 적용
blog_task = Task(
    description="Write a well-formatted blog post between 100-500 words",
    expected_output="A well-formatted blog post between 100-500 words",
    agent=blog_agent,
    guardrails=[
        validate_word_count,   # 첫 번째: 길이 검증
        validate_no_profanity, # 두 번째: 콘텐츠 검사
        format_output,         # 세 번째: 결과 포맷
    ],
    guardrail_max_retries=3
)

이 예시에서 가드레일은 순서대로 실행돼요. validate_word_count가 단어 수를 확인하고, validate_no_profanity가 1단계 결과를 받아 부적절한 언어를 검사하고, format_output이 2단계 결과를 받아 최종 포맷을 만져요. 만약 어느 단계에서든 실패하면 그 오류가 에이전트에게 전달되고, Task는 guardrail_max_retries 횟수만큼 다시 시도해요.

함수 기반과 LLM 기반 가드레일을 한 리스트에 섞는 것도 가능해요. 이러면 프로그램적 검증의 정밀도와, 주관적 기준에 대한 LLM 판단의 유연성을 함께 얻을 수 있어요.

from typing import Tuple, Any
from crewai import TaskOutput, Task

def validate_word_count(result: TaskOutput) -> Tuple[bool, Any]:
    word_count = len(result.raw.split())
    if word_count < 100:
        return (False, f"Content too short: {word_count} words. Need at least 100 words.")
    if word_count > 500:
        return (False, f"Content too long: {word_count} words. Maximum is 500 words.")
    return (True, result.raw)

# 함수 기반과 LLM 기반 가드레일 섞기
blog_task = Task(
    description="Write a well-formatted blog post between 100-500 words",
    expected_output="A well-formatted blog post between 100-500 words",
    agent=blog_agent,
    guardrails=[
        validate_word_count,  # 함수 기반: 정확한 단어 수 검사
        "The content must be engaging and suitable for a general audience",  # LLM 기반: 주관적 품질 검사
        "The writing style should be clear, concise, and free of technical jargon"  # LLM 기반: 스타일 검증
    ],
    guardrail_max_retries=3
)

가드레일 결과 처리¶

가드레일이 (False, error)를 돌려주면 이렇게 흘러가요.

오류가 에이전트에게 전달돼요.
에이전트가 문제를 고치려고 다시 시도해요.
이 과정을 반복하되, 가드레일이 (True, result)를 반환하거나 최대 재시도 횟수(guardrail_max_retries)에 도달할 때까지 계속해요.

재시도 처리가 붙은 예시를 볼게요.

from typing import Tuple, Any
from crewai import TaskOutput, Task

def validate_json_output(result: TaskOutput) -> Tuple[bool, Any]:
    """JSON 출력을 검증하고 파싱한다."""
    try:
        data = json.loads(result.raw)
        return (True, data)
    except json.JSONDecodeError as e:
        return (False, "Invalid JSON format")

task = Task(
    description="Generate a JSON report",
    expected_output="A valid JSON object",
    agent=analyst,
    guardrail=validate_json_output,
    guardrail_max_retries=3  # 재시도 횟수 제한
)

비동기 실행 (Asynchronous Execution)¶

async_execution=True로 설정하면 Task를 비동기로 실행해요. 쉽게 말해, 크루가 이 Task가 끝나기를 기다리지 않고 다음 Task로 넘어간다는 뜻이에요. 완료까지 오래 걸리거나, 이후 Task에 꼭 필요한 게 아닌 작업에 유용해요. 그리고 나중에 context 속성을 써서, 비동기 Task의 출력이 완료되기를 기다리게 하는 Task를 정의할 수 있어요.

# ...

list_ideas = Task(
    description="List of 5 interesting ideas to explore for an article about AI.",
    expected_output="Bullet point list of 5 ideas for an article.",
    agent=researcher,
    async_execution=True  # 비동기로 실행됨
)

list_important_history = Task(
    description="Research the history of AI and give me the 5 most important events.",
    expected_output="Bullet point list of 5 important events.",
    agent=researcher,
    async_execution=True  # 비동기로 실행됨
)

write_article = Task(
    description="Write an article about AI, its history, and interesting ideas.",
    expected_output="A 4 paragraph article about AI.",
    agent=writer,
    context=[list_ideas, list_important_history]  # 두 Task의 출력 완료를 기다림
)

# ...

콜백 메커니즘 (Callback Mechanism)¶

콜백은 Task가 완료된 뒤에 실행되는 함수예요. 작업 결과에 따라 알림을 보내거나 특정 액션을 트리거할 때 써요.

# ...

def callback_function(output):
    # Task 완료 후 할 일
    # 예: 매니저에게 이메일 보내기
    print(f"""
Task completed!
Task: {output.description}
Output: {output.raw}
""")

research_task = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool],
    callback=callback_function
)

# ...

특정 Task 출력에 접근하기¶

크루가 끝나면, 특정 Task의 출력은 그 Task 객체의 output 속성으로 접근할 수 있어요.

# ...

task1 = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool]
)

# ...

crew = Crew(
    agents=[research_agent],
    tasks=[task1, task2, task3],
    verbose=True
)
result = crew.kickoff()  # TaskOutput 객체를 반환

print(f"""
Task completed!
Task: {task1.output.description}
Output: {task1.output.raw}
""")

도구 오버라이드 (Tool Override Mechanism)¶

Task에 도구를 지정하면 에이전트가 가진 도구 능력을 동적으로 바꿀 수 있어요. 이게 CrewAI의 유연성을 보여 주는 부분이에요. 즉 에이전트 레벨이 아니라 Task 단위로 "이 작업에서는 이 도구만 써"라고 제한을 걸 수 있는 거죠.

오류 처리와 검증 메커니즘¶

Task를 만들고 실행하는 동안, 속성의 견고함과 신뢰성을 보장하기 위한 검증 장치들이 있어요. 대표적인 것만 보면 이래요.

Task마다 출력 타입을 하나만 설정하도록 해서, 출력 기대치를 명확하게 유지해요.
id 속성은 수동으로 할당하지 못하게 해서, 고유 식별자 체계의 무결성을 지켜요.

이런 검증들이 크루 실행의 일관성과 신뢰성을 유지해 줘요.

파일 저장 시 디렉터리 만들기¶

create_directory 파라미터는 Task 결과를 파일로 저장할 때, 디렉터리를 자동으로 만들어 줄지 결정해요. 복잡한 프로젝트 구조에서 출력물을 정리하고 파일 경로가 올바르게 잡히도록 하는 데 특히 유용해요.

기본 동작: 기본값은 create_directory=True라서, 출력 파일 경로에 없는 디렉터리가 있으면 자동으로 만들어요.

# 기본 동작 - 디렉터리가 자동 생성됨
report_task = Task(
    description='Generate a comprehensive market analysis report',
    expected_output='A detailed market analysis with charts and insights',
    agent=analyst_agent,
    output_file='reports/2025/market_analysis.md',  # 없으면 'reports/2025/'를 생성
    markdown=True
)

자동 생성 끄기: 디렉터리가 반드시 이미 존재해야 한다면 create_directory=False로 설정해요. 디렉터리가 없으면 RuntimeError를 발생시켜요.

# 엄격 모드 - 디렉터리가 이미 존재해야 함
strict_output_task = Task(
    description='Save critical data that requires existing infrastructure',
    expected_output='Data saved to pre-configured location',
    agent=data_agent,
    output_file='secure/vault/critical_data.json',
    create_directory=False  # 'secure/vault/'가 없으면 RuntimeError 발생
)

create_directory=False 상태에서 디렉터리가 없으면 CrewAI가 RuntimeError를 일으키는데, 코드에서 이렇게 잡아서 처리할 수 있어요.

try:
    result = crew.kickoff()
except RuntimeError as e:
    # 디렉터리 누락 처리
    print(f"Directory creation failed: {e}")
    # 직접 디렉터리를 만들거나 대체 경로를 사용

데이터스케쳐스 실무 관점¶

데이터스케쳐스에서 CrewAI 크루를 설계할 때 가장 먼저 잡아야 할 것은 출력(output)의 형태예요. 데이터 분석 워크플로는 최종 결과물이 표, 요약 보고서, 또는 특정 스키마를 가진 JSON 같은 구조화된 형태여야 하는 경우가 많거든요. 그럴 땐 Task에 output_pydantic이나 output_json으로 Pydantic 모델을 지정해 주면, 에이전트가 내뱉는 자유 텍스트를 우리가 원하는 형태로 강제할 수 있어요. 그리고 다음 Task로 넘어가기 전에 출력을 검증해야 할 때는 가드레일이 큰 힘을 발휘해요. 예를 들어 "이 보고서는 1000자 이상이고 출처가 5개 이상이어야 해" 같은 기준을 함수나 문자열로 걸어 두면, 품질이 낮은 결과물이 아래 단계로 새어 내려가는 걸 막을 수 있어요. 실제 파이프라인을 만들 땐 순차(sequential)로 시작해 실행 흐름을 눈으로 확인하고, 병렬이 필요한 지점만 비동기 실행(async_execution=True)과 context 조합으로 바꾸는 걸 추천해요. Task가 많아질수록 description과 expected_output을 명확히 쓰는 게 디버깅 시간을 크게 줄여 주는 것도 기억해 두면 좋아요. 참고로 최신 문법이나 추가 필드(response_model, converter_cls 등)는 문서 버전마다 조금씩 달라질 수 있으니, 배포 전에 공식 문서를 한 번 더 확인하는 걸 권장해요.

더 알아보기¶

원문: CrewAI 문서 — Tasks