CrewAI에서 Task 만들기와 관리 개요

CrewAI 프레임워크에서 Task를 생성하고 관리하는 방법에 대한 개요

Posted Apr 22, 2025 Updated Apr 24, 2025

By mminzy22

views 35 min read

Task란 무엇인가?

CrewAI 프레임워크에서 Task는 특정 Agent가 수행하는 구체적인 작업 단위입니다.

Task는 실행에 필요한 모든 정보를 담고 있으며, 설명(description), 담당 에이전트(agent), 필요한 도구(tools) 등 복잡한 작업 흐름을 지원하는 다양한 속성을 포함합니다.

CrewAI의 Task는 협업 형태로 구성될 수 있으며, 여러 에이전트가 함께 작업하는 구조를 지원합니다. 이러한 협업은 Task의 속성들과 Crew의 실행 방식(process)에 의해 조율됩니다.

Enterprise 기능: Visual Task Builder
CrewAI Enterprise 버전에서는 Crew Studio 내의 Visual Task Builder를 제공하여 복잡한 Task 생성과 연결을 쉽게 시각화할 수 있습니다.
Visual Task Builder의 기능:
드래그 앤 드롭 기반 Task 생성
Task 간 의존성 및 흐름 시각화
실시간 테스트 및 검증
팀 간 쉬운 공유 및 협업

Task 실행 방식

Task는 다음 두 가지 방식으로 실행할 수 있습니다:

Sequential(순차적): 정의된 순서대로 순차적으로 실행됨
Hierarchical(계층적): 에이전트의 역할 및 전문성에 따라 작업이 분배됨

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

Task 속성

속성명	파라미터 이름	타입	설명
설명	`description`	`str`	작업의 내용을 명확하고 간결하게 기술한 설명
예상 결과	`expected_output`	`str`	작업이 완료되었을 때의 기대 결과를 자세히 기술
이름 (옵션)	`name`	`Optional[str]`	작업의 식별을 위한 이름 지정
담당 에이전트 (옵션)	`agent`	`Optional[BaseAgent]`	해당 작업을 수행할 에이전트 지정
도구 (옵션)	`tools`	`List[BaseTool]`	해당 작업에서 사용할 수 있는 도구 제한
컨텍스트 (옵션)	`context`	`Optional[List["Task"]]`	해당 작업에 참조될 다른 작업들의 출력
비동기 실행 (옵션)	`async_execution`	`Optional[bool]`	비동기 실행 여부 (기본값은 False)
사람 검토 (옵션)	`human_input`	`Optional[bool]`	작업 결과에 대해 사람이 최종 검토해야 하는지 여부 (기본값 False)
설정값 (옵션)	`config`	`Optional[Dict[str, Any]]`	작업별 세부 설정
출력 파일 (옵션)	`output_file`	`Optional[str]`	작업 결과를 저장할 파일 경로
출력 JSON (옵션)	`output_json`	`Optional[Type[BaseModel]]`	JSON 출력을 위한 Pydantic 모델 지정
출력 Pydantic (옵션)	`output_pydantic`	`Optional[Type[BaseModel]]`	Pydantic 모델 기반의 구조화된 출력
콜백 함수 (옵션)	`callback`	`Optional[Any]`	작업 완료 후 실행될 콜백 함수

Task 생성 방법

CrewAI에서 Task를 생성하는 방법은 두 가지가 있습니다:

YAML 설정 파일 사용 (추천)
파이썬 코드로 직접 정의

YAML 설정 사용 (추천)

YAML 방식은 Task 정의를 더 깔끔하고 유지보수하기 쉬운 방식으로 제공하며, CrewAI 프로젝트 내에서의 사용을 권장합니다.

CrewAI 프로젝트를 생성한 후 src/latest_ai_development/config/tasks.yaml 파일을 열어 원하는 Task 요구사항에 맞게 수정하면 됩니다.

YAML 파일 내 변수 (예: {topic})는 kickoff() 실행 시 입력값으로 자동 대체됩니다.

  
crew.kickoff(inputs={'topic': 'AI Agents'})

YAML 예시 및 코드 연동 방법

아래는 YAML 파일을 사용해 Task를 구성하는 예시입니다:

  
tasks.yaml

research_task:
  description: >
    {topic}에 대해 철저히 조사하세요.
    현재 연도가 2025년임을 고려해 관련되고 흥미로운 정보를 수집하세요.
  expected_output: >
    {topic}에 대한 가장 관련성 높은 정보 10가지 항목의 리스트
  agent: researcher

reporting_task:
  description: >
    앞서 수집한 정보를 바탕으로 각 주제를 자세한 섹션으로 확장해 리포트를 작성하세요.
    리포트는 충분히 상세하며 모든 관련 정보를 포함해야 합니다.
  expected_output: >
    주요 주제를 섹션별로 정리한 마크다운 형식의 전체 리포트
    단, '```' 문법은 사용하지 말 것
  agent: reporting_analyst
  output_file: report.md

이 YAML 설정을 코드에서 사용하려면 CrewBase를 상속받는 클래스를 작성하면 됩니다:

  
# src/latest_ai_development/crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool

@CrewBase
class LatestAiDevelopmentCrew():
  """LatestAiDevelopment crew"""

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

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

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

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

  @crew
  def crew(self) -> Crew:
    return Crew(
      agents=[
        self.researcher(),
        self.reporting_analyst()
      ],
      tasks=[
        self.research_task(),
        self.reporting_task()
      ],
      process=Process.sequential
    )

⚠️ 주의: YAML 파일에서 사용한 이름 (agents.yaml, tasks.yaml의 항목명)은 Python 코드 내 메서드 이름과 일치해야 합니다.

직접 코드로 Task 정의하기 (대안 방식)

YAML 대신 Python 코드로 직접 Task를 정의할 수도 있습니다:

  
from crewai import Task

research_task = Task(
    description="""
        AI 에이전트에 대한 철저한 조사를 수행합니다.
        현재 연도가 2025년임을 고려해 흥미롭고 관련된 정보를 수집하세요.
    """,
    expected_output="""
        AI 에이전트에 대한 가장 중요한 정보 10가지 항목을 나열
    """,
    agent=researcher
)

reporting_task = Task(
    description="""
        수집한 내용을 바탕으로 각 주제를 리포트 형식으로 확장합니다.
        리포트는 충분히 자세해야 하며 관련 정보를 모두 포함해야 합니다.
    """,
    expected_output="""
        주요 주제를 섹션별로 구성한 전체 리포트. 마크다운 형식이지만 '```'은 포함하지 말 것.
    """,
    agent=reporting_analyst,
    output_file="report.md"
)

팁: 특정 에이전트를 지정하지 않으면 CrewAI의 hierarchical 실행 프로세스에 따라 역할 및 가용성 기반으로 자동 할당될 수 있습니다.

Task Output

효율적인 AI 작업 흐름을 만들기 위해서는 Task의 출력 결과를 이해하는 것이 매우 중요합니다. CrewAI는 TaskOutput 클래스를 통해 다양한 형식의 출력 결과를 구조화하여 처리할 수 있게 해줍니다.

Task의 출력은 TaskOutput 클래스로 캡슐화되며, 기본적으로 raw 출력만 포함합니다. 만약 output_pydantic 또는 output_json 속성을 설정했다면 Pydantic 객체나 JSON 딕셔너리 형태의 출력도 함께 포함됩니다.

TaskOutput 속성

속성명	파라미터	타입	설명
설명	`description`	`str`	Task에 대한 설명
요약	`summary`	`Optional[str]`	설명의 앞 10단어로 생성된 자동 요약
원본 출력	`raw`	`str`	기본 출력 형식. 가장 먼저 생성되는 결과
Pydantic 출력	`pydantic`	`Optional[BaseModel]`	구조화된 결과를 담는 Pydantic 모델 객체
JSON 출력	`json_dict`	`Optional[Dict[str, Any]]`	JSON 형태의 결과 딕셔너리
에이전트 정보	`agent`	`str`	해당 작업을 수행한 에이전트 이름
출력 형식	`output_format`	`OutputFormat`	출력 형식: RAW, JSON, Pydantic 중 하나 (기본값은 RAW)

TaskOutput 주요 메서드 및 속성

메서드/속성	설명
json	출력 형식이 JSON일 경우 문자열 형태로 반환
to_dict	JSON 또는 Pydantic 출력을 딕셔너리로 변환
str	문자열 출력 반환 (Pydantic → JSON → RAW 순서로 우선 적용됨)

TaskOutput 사용 예시

  
# 예시 Task 정의
task = Task(
    description='최신 AI 뉴스 조사 및 요약',
    expected_output='중요한 AI 뉴스 5개를 불릿 리스트로 정리',
    agent=research_agent,
    tools=[search_tool]
)

# Crew 실행
crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)

result = crew.kickoff()

# Task 결과 접근
task_output = task.output

print(f"작업 설명: {task_output.description}")
print(f"요약: {task_output.summary}")
print(f"원본 출력: {task_output.raw}")
if task_output.json_dict:
    print(json.dumps(task_output.json_dict, indent=2))
if task_output.pydantic:
    print(task_output.pydantic)

Task 간 의존성 및 Context 활용

하나의 Task가 다른 Task의 출력 결과에 의존하도록 만들 수 있습니다. 이때는 context 속성을 활용합니다.

  
research_task = Task(
    description="AI 최신 동향 조사",
    expected_output="최근 AI 기술 동향 리스트",
    agent=researcher
)

analysis_task = Task(
    description="조사 결과 분석 및 주요 트렌드 도출",
    expected_output="AI 트렌드 분석 리포트",
    agent=analyst,
    context=[research_task]  # research_task의 결과를 context로 사용
)

Task Guardrails (출력 검증 및 후처리)

Task Guardrail 기능을 통해 다음 Task로 결과를 전달하기 전에 유효성 검사를 수행할 수 있습니다. 출력의 품질을 보장하고, 조건을 만족하지 않으면 에이전트가 다시 작업하도록 피드백을 제공합니다.

Guardrail 함수 사용 예시

  
from typing import Tuple, Any
from crewai import TaskOutput

def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    """블로그 콘텐츠가 200단어 이하인지 검증"""
    try:
        word_count = len(result.split())
        if word_count > 200:
            return (False, "블로그 글이 200단어를 초과했습니다")
        return (True, result.strip())
    except Exception as e:
        return (False, f"검증 중 오류 발생: {str(e)}")

blog_task = Task(
    description="AI에 대한 블로그 글 작성",
    expected_output="200단어 이하의 블로그 글",
    agent=blog_agent,
    guardrail=validate_blog_content  # guardrail 함수 지정
)

Guardrail 함수는 (bool, Any) 형태의 튜플을 반환해야 하며, True면 유효성 통과, False면 실패 메시지를 반환합니다. 실패 시 에이전트는 수정된 응답을 다시 생성하려 시도하게 됩니다.

Guardrail 함수 요구 사항

함수 시그니처:
- 파라미터는 하나만 받아야 하며 (예: result: TaskOutput)
- 반환 값은 (bool, Any) 형태의 튜플이어야 합니다
- 타입 힌트는 권장 사항입니다 (선택 사항)
반환 값 기준:
- 성공 시: (True, 결과값) 형태 (예: (True, validated_result))
- 실패 시: (False, 에러 메시지) 형태 (예: (False, "출력 형식 오류"))

에러 처리 베스트 프랙티스

구조화된 에러 응답 예시:

  
from crewai import TaskOutput

def validate_with_context(result: TaskOutput) -> Tuple[bool, Any]:
    try:
        validated_data = perform_validation(result)
        return (True, validated_data)
    except ValidationError as e:
        return (False, f"VALIDATION_ERROR: {str(e)}")
    except Exception as e:
        return (False, str(e))

에러 분류 팁:
- 특정 에러 코드 사용
- 관련된 컨텍스트 제공
- 명확한 피드백 메시지 포함
유효성 검증 체이닝 예시:

  
from typing import Any, Tuple
from crewai import TaskOutput

def complex_validation(result: TaskOutput) -> Tuple[bool, Any]:
    # 1단계: 기본 검사
    if not result:
        return (False, "출력이 비어 있습니다")

    # 2단계: 내용 검사
    try:
        validated = validate_content(result)
        if not validated:
            return (False, "내용이 유효하지 않습니다")

        # 3단계: 형식 변환
        formatted = format_output(validated)
        return (True, formatted)
    except Exception as e:
        return (False, str(e))

Guardrail 결과 처리 흐름

Guardrail 함수가 (False, 오류 메시지)를 반환하면:

오류 메시지는 해당 에이전트에게 전달됩니다
에이전트는 문제를 수정한 새로운 응답을 생성하려고 시도합니다
다음 두 조건 중 하나가 충족될 때까지 반복됩니다:
- Guardrail이 (True, 결과)를 반환함
- 최대 재시도 횟수(max_retries)에 도달함

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

def validate_json_output(result: TaskOutput) -> Tuple[bool, Any]:
    try:
        data = json.loads(result)
        return (True, data)
    except json.JSONDecodeError:
        return (False, "유효한 JSON 형식이 아닙니다")

task = Task(
    description="JSON 리포트 생성",
    expected_output="유효한 JSON 객체",
    agent=analyst,
    guardrail=validate_json_output,
    max_retries=3  # 최대 재시도 횟수 지정
)

일관된 구조화된 출력 받기

Crew의 마지막 Task 출력은 전체 Crew의 최종 출력으로 간주됩니다.

작업 출력이 일정한 구조를 갖도록 하기 위해 output_pydantic 또는 output_json 속성을 사용하는 것이 중요합니다. 이 설정을 통해 출력의 일관성과 데이터 활용성을 크게 높일 수 있습니다.

일관된 구조화된 출력 받기

Crew의 마지막 Task 출력은 전체 Crew의 최종 출력으로 간주됩니다.

`output_pydantic` 사용하기

output_pydantic 속성을 사용하면 작업 결과가 특정 Pydantic 모델을 따르도록 강제할 수 있습니다. 이는 출력의 구조화뿐만 아니라 유효성 검사까지 보장해줍니다.

  
import json

from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

class Blog(BaseModel):
    title: str
    content: str

blog_agent = Agent(
    role="Blog Content Generator Agent",
    goal="블로그 제목과 내용을 생성합니다",
    backstory="""당신은 흥미롭고 유익한 블로그 콘텐츠를 제작하는 전문가입니다.""",
    verbose=False,
    allow_delegation=False,
    llm="gpt-4o",
)

task1 = Task(
    description="""주어진 주제에 대해 블로그 제목과 본문을 작성하세요. 200단어 이내로 작성해야 합니다.""",
    expected_output="설득력 있는 블로그 제목과 잘 정리된 본문",
    agent=blog_agent,
    output_pydantic=Blog,
)

crew = Crew(
    agents=[blog_agent],
    tasks=[task1],
    verbose=True,
    process=Process.sequential,
)

result = crew.kickoff()

# 방식 1: 딕셔너리 스타일 인덱싱
print("[방법 1] 딕셔너리 인덱싱")
title = result["title"]
content = result["content"]
print("Title:", title)
print("Content:", content)

# 방식 2: Pydantic 객체에서 직접 접근
print("[방법 2] Pydantic 객체 직접 접근")
title = result.pydantic.title
content = result.pydantic.content
print("Title:", title)
print("Content:", content)

# 방식 3: to_dict() 메서드 사용
print("[방법 3] to_dict()로 딕셔너리 변환")
output_dict = result.to_dict()
title = output_dict["title"]
content = output_dict["content"]
print("Title:", title)
print("Content:", content)

# 방식 4: 전체 객체 출력
print("[방법 4] 전체 출력")
print("Blog:", result)

이 예시에서는 다음과 같은 과정을 거칩니다:

Blog라는 Pydantic 모델을 정의하고 title, content 필드를 포함함
task1에서 output_pydantic=Blog를 지정하여 결과가 해당 모델 구조를 따르도록 설정
Crew 실행 후 여러 방식으로 구조화된 결과에 접근 가능

출력 접근 방식 요약

딕셔너리 스타일 인덱싱: result["필드명"]으로 접근 (CrewOutput 클래스가 __getitem__을 구현함)
Pydantic 객체에서 직접 접근: result.pydantic.필드명으로 직접 접근 가능
to_dict() 사용: .to_dict()로 전체 결과를 딕셔너리로 변환
전체 객체 출력: print(result)로 전체 객체를 문자열 형태로 출력

`output_json` 사용하기

output_json 속성을 사용하면 작업의 출력이 JSON 형식을 갖추도록 강제할 수 있습니다. 이 방식은 애플리케이션 내에서 쉽게 파싱하고 사용할 수 있도록 도와줍니다.

  
import json

from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

# Pydantic 모델 정의
class Blog(BaseModel):
    title: str
    content: str

# 에이전트 정의
blog_agent = Agent(
    role="Blog Content Generator Agent",
    goal="블로그 제목과 본문 생성",
    backstory="""당신은 매력적인 블로그 콘텐츠를 작성하는 콘텐츠 제작 전문가입니다.""",
    verbose=False,
    allow_delegation=False,
    llm="gpt-4o",
)

# JSON 형식 출력을 지정한 Task
task1 = Task(
    description="""주어진 주제에 대한 블로그 제목과 내용을 생성하세요. 200단어 이하로 작성해야 합니다.""",
    expected_output="'title'과 'content' 필드를 포함한 JSON 객체",
    agent=blog_agent,
    output_json=Blog,
)

crew = Crew(
    agents=[blog_agent],
    tasks=[task1],
    verbose=True,
    process=Process.sequential,
)

result = crew.kickoff()

# 방식 1: 딕셔너리 인덱싱 접근
print("[방법 1] 딕셔너리 인덱싱")
title = result["title"]
content = result["content"]
print("Title:", title)
print("Content:", content)

# 방식 2: 전체 객체 출력
print("[방법 2] 전체 출력")
print("Blog:", result)

이 예시에서는 다음과 같은 특징이 있습니다:

Blog라는 Pydantic 모델을 JSON 출력 구조로 정의합니다.
task1에서 output_json=Blog를 사용해 출력이 JSON 구조를 따르도록 설정합니다.
작업 완료 후 구조화된 JSON 결과를 딕셔너리처럼 혹은 전체 출력 형태로 접근할 수 있습니다.

출력 접근 방식 요약

딕셔너리 인덱싱: result["필드명"] 방식으로 접근 가능. CrewOutput 클래스가 __getitem__을 구현함
전체 객체 출력: print(result)는 JSON 형식의 문자열로 출력됨

output_pydantic 또는 output_json 속성을 활용하면 Task 결과를 구조화된 형식으로 일관되게 다룰 수 있어, 애플리케이션 전반에서 데이터를 활용하거나 후속 작업(Task)으로 넘기는 데에 큰 도움이 됩니다.

Task에 도구(Tool) 통합하기

crewAI-tools 또는 LangChain Tools를 통해 에이전트의 작업 성능을 향상시킬 수 있습니다.

Tool을 사용하는 Task 예시

  
import os
os.environ["OPENAI_API_KEY"] = "Your Key"
os.environ["SERPER_API_KEY"] = "Your Key"  # serper.dev API 키

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool

research_agent = Agent(
  role='Researcher',
  goal='최신 AI 뉴스를 조사하고 요약합니다',
  backstory="""당신은 대기업에서 데이터를 분석하고 통찰을 제공하는 리서처입니다.""",
  verbose=True
)

# 웹 검색을 위한 도구 정의
search_tool = SerperDevTool()

task = Task(
  description='최신 AI 뉴스 조사 및 요약',
  expected_output='중요한 AI 뉴스 5개를 불릿 리스트로 요약',
  agent=research_agent,
  tools=[search_tool]  # 이 Task에만 이 Tool을 사용
)

crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)

result = crew.kickoff()
print(result)

이 예시처럼 특정 Task에만 Tool을 지정할 수 있으며, 이는 해당 작업에 가장 적합한 도구로 실행되도록 유연하게 설정할 수 있음을 보여줍니다.

다른 Task의 출력 참조하기

CrewAI에서는 기본적으로 앞선 Task의 출력이 다음 Task로 전달되지만, 특정 Task의 출력을 명시적으로 참조할 수도 있습니다. context 속성을 사용하면 여러 Task의 결과를 다음 Task에 입력으로 전달할 수 있습니다.

이는 특정 Task가 바로 직전 Task가 아닌 다른 Task의 결과를 필요로 할 때 유용합니다.

  
research_ai_task = Task(
    description="최신 AI 기술 동향 조사",
    expected_output="AI 최신 동향 리스트",
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

research_ops_task = Task(
    description="최신 AI Ops 동향 조사",
    expected_output="AI Ops 최신 동향 리스트",
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

write_blog_task = Task(
    description="AI의 중요성과 최근 동향을 주제로 한 블로그 포스트 작성",
    expected_output="4단락으로 구성된 블로그 포스트",
    agent=writer_agent,
    context=[research_ai_task, research_ops_task]
)

비동기 Task 실행

Task에 async_execution=True 옵션을 주면 비동기 실행이 가능해집니다. Crew는 이 Task의 완료 여부와 상관없이 다음 Task로 넘어갑니다. 이후 특정 Task에서 context로 해당 비동기 Task를 지정해 결과가 준비될 때까지 대기하도록 만들 수 있습니다.

  
list_ideas = Task(
    description="AI 주제에 대한 흥미로운 아이디어 5가지 나열",
    expected_output="불릿 리스트 5개",
    agent=researcher,
    async_execution=True
)

list_important_history = Task(
    description="AI의 역사에서 가장 중요한 사건 5가지 조사",
    expected_output="불릿 리스트 5개",
    agent=researcher,
    async_execution=True
)

write_article = Task(
    description="AI의 역사와 주요 아이디어를 포함한 4단락 기사 작성",
    expected_output="4단락 기사",
    agent=writer,
    context=[list_ideas, list_important_history]
)

Callback 기능

작업이 완료된 후 자동으로 실행되는 후처리 로직을 callback 속성으로 정의할 수 있습니다. 예를 들어 작업 결과를 관리자에게 이메일로 보내는 작업 등입니다.

  
def callback_function(output: TaskOutput):
    print(f"""
        Task 완료!
        작업 설명: {output.description}
        결과:
        {output.raw}
    """)

research_task = Task(
    description='최신 AI 뉴스 조사 및 요약',
    expected_output='최신 AI 뉴스 5가지 불릿 요약',
    agent=research_agent,
    tools=[search_tool],
    callback=callback_function
)

특정 Task 출력 접근하기

Crew 실행이 완료되면 개별 Task 객체의 output 속성을 통해 해당 작업의 결과에 접근할 수 있습니다.

  
task1 = Task(
    description='최신 AI 뉴스 조사 및 요약',
    expected_output='최신 AI 뉴스 5가지 불릿 요약',
    agent=research_agent,
    tools=[search_tool]
)

crew = Crew(
    agents=[research_agent],
    tasks=[task1, task2, task3],
    verbose=True
)

result = crew.kickoff()

print(f"""
    작업 완료!
    설명: {task1.output.description}
    출력:
    {task1.output.raw}
""")

도구 재정의(Override) 메커니즘

Task 레벨에서 tools를 지정하면 에이전트의 기본 도구 구성과 상관없이 해당 Task 전용 도구로 작업을 수행할 수 있습니다. 이는 CrewAI의 유연성을 잘 보여주는 기능입니다.

에러 처리 및 유효성 검사 메커니즘

CrewAI는 Task를 생성하고 실행하는 과정에서 일관성 있고 신뢰성 있는 실행을 보장하기 위해 여러 유효성 검사를 수행합니다. 주요 예시는 다음과 같습니다:

하나의 Task에 대해 하나의 출력 타입만 설정하도록 제한 (raw, json, pydantic 중 하나)
Task의 id 속성은 수동으로 지정할 수 없으며, 내부에서 자동으로 고유하게 생성됨

이러한 검증은 전체 실행 흐름의 안정성과 예측 가능성을 높여줍니다.

Task Guardrails 다시 보기

Guardrail은 Task의 출력 결과를 후속 Task로 넘기기 전에 검증·필터링·변환할 수 있는 기능입니다. Guardrail은 선택적 속성이며, 다음 Task가 실행되기 전에 자동으로 호출됩니다.

기본 사용 예시

  
from typing import Tuple, Union
from crewai import Task

def validate_json_output(result: str) -> Tuple[bool, Union[dict, str]]:
    try:
        json_data = json.loads(result)
        return (True, json_data)
    except json.JSONDecodeError:
        return (False, "JSON 형식이 아닙니다")

task = Task(
    description="JSON 데이터 생성",
    expected_output="유효한 JSON 객체",
    guardrail=validate_json_output
)

Guardrail 동작 방식

선택적 속성: 필요한 Task에만 Guardrail을 설정 가능
실행 시점: 다음 Task가 시작되기 전에 실행됨
반환 형식: (성공 여부, 결과 또는 에러 메시지) 형식의 튜플 반환
흐름 처리:
- 성공 시: 다음 Task로 결과 전달
- 실패 시: 에이전트가 결과를 재생성함

주요 사용 사례

데이터 형식 유효성 검사

  
def validate_email_format(result: str) -> Tuple[bool, Union[str, str]]:
    import re
    pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    if re.match(pattern, result.strip()):
        return (True, result.strip())
    return (False, "이메일 형식이 아닙니다")

민감 정보 필터링

  
def filter_sensitive_info(result: str) -> Tuple[bool, Union[str, str]]:
    patterns = ['SSN:', 'password:', 'secret:']
    for p in patterns:
        if p.lower() in result.lower():
            return (False, f"민감 정보 포함: {p}")
    return (True, result)

데이터 변환

  
def normalize_phone_number(result: str) -> Tuple[bool, Union[str, str]]:
    import re
    digits = re.sub(r'\D', '', result)
    if len(digits) == 10:
        formatted = f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
        return (True, formatted)
    return (False, "전화번호는 10자리여야 합니다")

고급 사용 방식

다중 검증 체이닝

  
def chain_validations(*validators):
    def combined(result):
        for v in validators:
            success, data = v(result)
            if not success:
                return (False, data)
            result = data
        return (True, result)
    return combined

task = Task(
    description="연락처 수집",
    expected_output="이메일과 전화번호",
    guardrail=chain_validations(
        validate_email_format,
        filter_sensitive_info
    )
)

사용자 정의 재시도 로직

  
task = Task(
    description="데이터 생성",
    expected_output="유효한 데이터",
    guardrail=validate_data,
    max_retries=5
)

출력 파일 저장 시 디렉토리 자동 생성

Task의 output_file 경로에 디렉토리가 포함되어 있고, 해당 디렉토리가 존재하지 않는 경우 자동으로 생성되도록 설정할 수 있습니다. 이때 create_directory=True 속성을 사용합니다.

  
save_output_task = Task(
    description='요약된 AI 뉴스를 파일에 저장',
    expected_output='파일 저장 성공',
    agent=research_agent,
    tools=[file_save_tool],
    output_file='outputs/ai_news_summary.txt',
    create_directory=True
)

결론

Task는 CrewAI 에이전트의 행동을 구성하는 핵심 요소입니다. 각 Task를 명확히 정의하고, 도구를 적절히 연결하며, 실행 흐름을 제어하고, Guardrail로 출력 품질을 보장하는 구조를 갖추면 보다 강력하고 신뢰할 수 있는 AI 시스템을 구현할 수 있습니다.

CrewAI

This post is licensed under CC BY 4.0 by the author.

Task란 무엇인가?

Task 실행 방식

Task 속성

Task 생성 방법

YAML 설정 사용 (추천)

YAML 예시 및 코드 연동 방법

직접 코드로 Task 정의하기 (대안 방식)

Task Output

TaskOutput 속성

TaskOutput 주요 메서드 및 속성

TaskOutput 사용 예시

Task 간 의존성 및 Context 활용

Task Guardrails (출력 검증 및 후처리)

Guardrail 함수 사용 예시

Guardrail 함수 요구 사항

에러 처리 베스트 프랙티스

Guardrail 결과 처리 흐름

일관된 구조화된 출력 받기

일관된 구조화된 출력 받기

output_pydantic 사용하기

출력 접근 방식 요약

output_json 사용하기

출력 접근 방식 요약

Task에 도구(Tool) 통합하기

Tool을 사용하는 Task 예시

다른 Task의 출력 참조하기

비동기 Task 실행

Callback 기능

특정 Task 출력 접근하기

도구 재정의(Override) 메커니즘

에러 처리 및 유효성 검사 메커니즘

Task Guardrails 다시 보기

기본 사용 예시

Guardrail 동작 방식

주요 사용 사례

데이터 형식 유효성 검사

민감 정보 필터링

데이터 변환

고급 사용 방식

다중 검증 체이닝

사용자 정의 재시도 로직

출력 파일 저장 시 디렉토리 자동 생성

결론

Trending Tags

`output_pydantic` 사용하기

`output_json` 사용하기