​
Task 개요

CrewAI 프레임워크에서 Task는 Agent가 수행하는 특정 작업을 의미한다.

Task는 설명, 담당 Agent, 필요한 도구 등 실행에 필요한 모든 세부 정보를 제공한다. 이를 통해 다양한 복잡도의 작업을 처리할 수 있다.

CrewAI의 Task는 여러 Agent가 협력해야 하는 경우도 있다. Task 속성을 통해 이를 관리하며, Crew의 프로세스가 조율하여 팀워크와 효율성을 높인다.

​
작업 실행 흐름

작업은 두 가지 방식으로 실행할 수 있다:

  • 순차적: 정의된 순서대로 작업을 실행
  • 계층적: 에이전트의 역할과 전문성에 따라 작업을 할당

실행 흐름은 크루를 생성할 때 정의한다:

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

​
태스크 속성

속성파라미터타입설명
설명descriptionstr태스크의 내용을 명확하고 간결하게 설명한다.
예상 출력expected_outputstr태스크 완료 시 기대하는 결과를 상세히 기술한다.
이름 (선택)nameOptional[str]태스크를 식별하기 위한 이름이다.
에이전트 (선택)agentOptional[BaseAgent]태스크 실행을 담당하는 에이전트이다.
도구 (선택)toolsList[BaseTool]태스크 수행에 사용할 수 있는 도구나 리소스를 제한한다.
컨텍스트 (선택)contextOptional[List["Task"]]이 태스크에 컨텍스트로 사용될 다른 태스크의 출력이다.
비동기 실행 (선택)async_executionOptional[bool]태스크를 비동기적으로 실행할지 여부를 결정한다. 기본값은 False이다.
사람 입력 (선택)human_inputOptional[bool]에이전트의 최종 답변을 사람이 검토할지 여부를 결정한다. 기본값은 False이다.
설정 (선택)configOptional[Dict[str, Any]]태스크별 설정 파라미터이다.
출력 파일 (선택)output_fileOptional[str]태스크 출력을 저장할 파일 경로이다.
출력 JSON (선택)output_jsonOptional[Type[BaseModel]]JSON 출력을 구조화하기 위한 Pydantic 모델이다.
출력 Pydantic (선택)output_pydanticOptional[Type[BaseModel]]태스크 출력을 위한 Pydantic 모델이다.
콜백 (선택)callbackOptional[Any]태스크 완료 후 실행할 함수나 객체이다.

​
작업 생성하기

CrewAI에서 작업을 생성하는 방법은 두 가지가 있다: YAML 설정 파일 사용(권장) 또는 코드에서 직접 정의하는 방법이다.

​
YAML 설정 (권장)

YAML 설정을 사용하면 작업을 정의하는 더 깔끔하고 유지보수하기 쉬운 방법을 제공한다. CrewAI 프로젝트에서 작업을 정의할 때 이 방법을 적극 권장한다.

설치 섹션에서 설명한 대로 CrewAI 프로젝트를 생성한 후, src/latest_ai_development/config/tasks.yaml 파일로 이동해 템플릿을 수정하여 특정 작업 요구사항에 맞게 조정한다.

YAML 파일의 변수(예: {topic})는 crew를 실행할 때 입력값으로 대체된다:

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

다음은 YAML을 사용해 작업을 구성하는 예제이다:

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를 상속받는 crew 클래스를 생성한다:

crew.py
# 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'],
      verbose=True,
      tools=[SerperDevTool()]
    )

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

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

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

  @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 코드의 메서드 이름과 일치해야 한다.

​
직접 코드로 정의하기 (대안)

YAML 설정 없이 코드에서 직접 작업을 정의할 수도 있다:

task.py
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"
)

agent를 직접 지정하거나, hierarchical CrewAI 프로세스가 역할, 가용성 등을 기반으로 결정하도록 할 수 있다.

​
작업 출력 이해하기

작업 출력을 이해하는 것은 효과적인 AI 워크플로우를 구축하는 데 중요하다. CrewAI는 TaskOutput 클래스를 통해 작업 결과를 구조화된 방식으로 처리하며, 다양한 출력 포맷을 지원하고 작업 간에 쉽게 전달할 수 있다.

CrewAI 프레임워크에서 작업의 출력은 TaskOutput 클래스로 캡슐화된다. 이 클래스는 작업 결과에 접근할 수 있는 구조화된 방식을 제공하며, raw 출력, JSON, Pydantic 모델 등 다양한 포맷을 지원한다.

기본적으로 TaskOutput은 raw 출력만 포함한다. pydantic 또는 json_dict 출력은 원래 Task 객체가 output_pydantic 또는 output_json으로 구성된 경우에만 포함된다.

​
작업 출력 속성

속성매개변수타입설명
설명descriptionstr작업에 대한 설명
요약summaryOptional[str]작업의 요약, 설명의 첫 10단어에서 자동 생성
원본rawstr작업의 원본 출력. 이는 출력의 기본 형식
PydanticpydanticOptional[BaseModel]작업의 구조화된 출력을 나타내는 Pydantic 모델 객체
JSON 딕셔너리json_dictOptional[Dict[str, Any]]작업의 JSON 출력을 나타내는 딕셔너리
에이전트agentstr작업을 실행한 에이전트
출력 형식output_formatOutputFormat작업 출력의 형식. RAW, JSON, Pydantic 중 선택 가능. 기본값은 RAW

​
작업 메서드와 속성

메서드/속성설명
json출력 형식이 JSON일 경우, 작업 출력의 JSON 문자열 표현을 반환한다.
to_dictJSON 및 Pydantic 출력을 딕셔너리로 변환한다.
str작업 출력의 문자열 표현을 반환한다. Pydantic, JSON, 원시 데이터 순으로 우선순위를 둔다.

​
태스크 출력에 접근하기

태스크가 실행된 후, Task 객체의 output 속성을 통해 출력 결과에 접근할 수 있다. TaskOutput 클래스는 이 출력 결과를 다루고 표현하는 다양한 방법을 제공한다.

​
예제

# 예제 작업
task = Task(
    description='최신 AI 뉴스를 찾아 요약하기',
    expected_output='가장 중요한 AI 뉴스 5개를 요약한 목록',
    agent=research_agent,
    tools=[search_tool]
)

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

result = crew.kickoff()

# 작업 결과 접근
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(f"JSON 출력: {json.dumps(task_output.json_dict, indent=2)}")
if task_output.pydantic:
    print(f"Pydantic 출력: {task_output.pydantic}")

​
작업 의존성과 컨텍스트

작업은 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가 완료될 때까지 대기
)

​
태스크 가드레일

태스크 가드레일은 한 태스크의 출력을 다음 태스크로 전달하기 전에 검증하고 변환하는 방법을 제공한다. 이 기능은 데이터 품질을 보장하고, 에이전트의 출력이 특정 기준을 충족하지 못할 때 피드백을 제공하는 데 유용하다.

​
태스크 가드레일 사용하기

태스크에 가드레일을 추가하려면 guardrail 파라미터를 통해 검증 함수를 제공한다:

from typing import Tuple, Union, Dict, Any

def validate_blog_content(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
    """블로그 콘텐츠가 요구 사항을 충족하는지 검증한다."""
    try:
        # 단어 수 확인
        word_count = len(result.split())
        if word_count > 200:
            return (False, {
                "error": "블로그 콘텐츠가 200단어를 초과함",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })

        # 추가 검증 로직
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "검증 중 예상치 못한 오류 발생",
            "code": "SYSTEM_ERROR"
        })

blog_task = Task(
    description="AI에 관한 블로그 포스트 작성",
    expected_output="200단어 이하의 블로그 포스트",
    agent=blog_agent,
    guardrail=validate_blog_content  # 가드레일 함수 추가
)

​
가드레일 기능 요구사항

  1. 함수 시그니처:

    • 반드시 하나의 파라미터만 받아야 함 (태스크 출력값)
    • (bool, Any) 형태의 튜플을 반환해야 함
    • 타입 힌트는 권장되지만 필수는 아님

  2. 반환값:

    • 성공 시: (True, 검증된_결과) 반환
    • 실패 시: (False, 오류_상세정보) 반환

​
에러 처리 모범 사례

  1. 구조화된 에러 응답:
def validate_with_context(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
    try:
        # 주요 검증 로직
        validated_data = perform_validation(result)
        return (True, validated_data)
    except ValidationError as e:
        return (False, {
            "error": str(e),
            "code": "VALIDATION_ERROR",
            "context": {"input": result}
        })
    except Exception as e:
        return (False, {
            "error": "예상치 못한 에러",
            "code": "SYSTEM_ERROR"
        })

  1. 에러 카테고리:

    • 구체적인 에러 코드 사용
    • 관련 컨텍스트 포함
    • 실행 가능한 피드백 제공

  2. 검증 체인:

from typing import Any, Dict, List, Tuple, Union

def complex_validation(result: str) -> Tuple[bool, Union[str, Dict[str, Any]]]:
    """다중 검증 단계를 체이닝한다."""
    # 단계 1: 기본 검증
    if not result:
        return (False, {"error": "빈 결과값", "code": "EMPTY_INPUT"})

    # 단계 2: 내용 검증
    try:
        validated = validate_content(result)
        if not validated:
            return (False, {"error": "유효하지 않은 내용", "code": "CONTENT_ERROR"})

        # 단계 3: 형식 검증
        formatted = format_output(validated)
        return (True, formatted)
    except Exception as e:
        return (False, {
            "error": str(e),
            "code": "VALIDATION_ERROR",
            "context": {"step": "content_validation"}
        })

​
가드레일 결과 처리

가드레일이 (False, error)를 반환하면:

  1. 에러를 에이전트에게 전달한다
  2. 에이전트가 문제를 해결하려고 시도한다
  3. 다음 중 하나가 발생할 때까지 이 과정을 반복한다:
    • 가드레일이 (True, result)를 반환한다
    • 최대 재시도 횟수에 도달한다

재시도 처리를 포함한 예제:

from typing import Optional, Tuple, Union

def validate_json_output(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
    """JSON 출력을 검증하고 파싱한다."""
    try:
        # JSON으로 파싱 시도
        data = json.loads(result)
        return (True, data)
    except json.JSONDecodeError as e:
        return (False, {
            "error": "잘못된 JSON 형식",
            "code": "JSON_ERROR",
            "context": {"line": e.lineno, "column": e.colno}
        })

task = Task(
    description="JSON 보고서 생성",
    expected_output="유효한 JSON 객체",
    agent=analyst,
    guardrail=validate_json_output,
    max_retries=3  # 최대 재시도 횟수 제한
)

​
구조화된 일관된 출력 얻기

크루의 최종 작업 결과는 크루 자체의 최종 출력이 된다는 점을 주목해야 한다.

​
output_pydantic 사용하기

output_pydantic 속성을 사용하면 작업의 출력이 준수해야 하는 Pydantic 모델을 정의할 수 있다. 이를 통해 출력이 구조화될 뿐만 아니라 Pydantic 모델에 따라 검증된다.

다음은 output_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="블로그 콘텐츠 생성 에이전트",
    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)
print("내용:", content)

# 옵션 2: Pydantic 모델에서 직접 속성 접근
print("속성 접근 - 옵션 2")
title = result.pydantic.title
content = result.pydantic.content
print("제목:", title)
print("내용:", content)

# 옵션 3: to_dict() 메서드를 사용해 속성 접근
print("속성 접근 - 옵션 3")
output_dict = result.to_dict()
title = output_dict["title"]
content = output_dict["content"]
print("제목:", title)
print("내용:", content)

# 옵션 4: 전체 블로그 객체 출력
print("속성 접근 - 옵션 5")
print("블로그:", result)

이 예제에서:

  • title과 content 필드를 가진 Pydantic 모델 Blog를 정의했다.
  • task1 작업은 output_pydantic 속성을 사용해 출력이 Blog 모델을 준수하도록 지정했다.
  • 크루를 실행한 후, 여러 가지 방법으로 구조화된 출력에 접근할 수 있다.

​
출력값에 접근하는 방법 설명

  1. 딕셔너리 스타일 인덱싱: result["field_name"]을 사용해 필드에 직접 접근할 수 있다. 이는 CrewOutput 클래스가 __getitem__ 메서드를 구현했기 때문에 가능하다.
  2. Pydantic 모델에서 직접 접근: result.pydantic 객체에서 속성을 직접 접근한다.
  3. to_dict() 메서드 사용: 출력값을 딕셔너리로 변환한 후 필드에 접근한다.
  4. 전체 객체 출력: result 객체를 그대로 출력해 구조화된 결과를 확인한다.

​
output_json 사용하기

output_json 속성을 사용하면 JSON 형식으로 예상 출력을 정의할 수 있다. 이를 통해 작업의 출력이 유효한 JSON 구조로 반환되도록 보장하며, 애플리케이션에서 쉽게 파싱하고 활용할 수 있다.

다음은 output_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="블로그 콘텐츠 생성 에이전트",
    goal="블로그 제목과 내용 생성",
    backstory="""당신은 매력적이고 유익한 블로그 포스트를 작성하는 데 능숙한 전문 콘텐츠 크리에이터다.""",
    verbose=False,
    allow_delegation=False,
    llm="gpt-4o",
)

# output_json을 Blog 모델로 설정하여 작업 정의
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)
print("내용:", content)

# 옵션 2: 전체 블로그 객체 출력
print("속성 접근 - 옵션 2")
print("블로그:", result)

이 예제에서:

  • title과 content 필드를 가진 Pydantic 모델 Blog를 정의하여 JSON 출력의 구조를 명시한다.
  • 작업 task1은 output_json 속성을 사용해 Blog 모델에 맞는 JSON 출력을 기대함을 나타낸다.
  • 크루를 실행한 후, 구조화된 JSON 출력에 접근하는 두 가지 방법을 보여준다.

​
출력에 접근하는 방법 설명

  1. 딕셔너리 스타일 인덱싱을 사용해 속성에 접근하기: result["field_name"]과 같은 방식으로 필드에 직접 접근할 수 있다. 이는 CrewOutput 클래스가 __getitem__ 메서드를 구현하고 있어 출력을 딕셔너리처럼 다룰 수 있기 때문이다. 이 옵션에서는 title과 content를 result에서 가져온다.
  2. 전체 블로그 객체 출력하기: result를 출력하면 CrewOutput 객체의 문자열 표현을 얻을 수 있다. __str__ 메서드가 JSON 출력을 반환하도록 구현되어 있기 때문에, 이는 블로그 객체를 나타내는 포맷된 문자열로 전체 출력을 보여준다.

output_pydantic 또는 output_json을 사용하면 여러분의 작업이 일관되고 구조화된 형식으로 출력을 생성하도록 보장할 수 있다. 이를 통해 애플리케이션 내부나 여러 작업 간에 데이터를 더 쉽게 처리하고 활용할 수 있다.

​
도구와 작업 통합하기

CrewAI Toolkit과 LangChain Tools에서 제공하는 도구를 활용해 작업 성능과 에이전트 상호작용을 강화한다.

​
도구를 사용하여 작업 생성하기

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='Find and summarize the latest AI news',
  backstory="""You're a researcher at a large company.
  You're responsible for analyzing data and providing insights
  to the business.""",
  verbose=True
)

# 인터넷에서 특정 쿼리에 대한 의미론적 검색을 수행하는 도구
search_tool = SerperDevTool()

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()
print(result)

이 예제는 특정 도구를 사용하여 작업을 생성할 때, 에이전트의 기본 설정을 재정의하여 맞춤형 작업 실행을 보여준다.

​
다른 태스크 참조하기

CrewAI에서는 한 태스크의 출력이 자동으로 다음 태스크로 전달된다. 하지만 특정 태스크의 출력을 다른 태스크의 컨텍스트로 사용하고 싶을 때, 여러 태스크의 출력을 명시적으로 정의할 수 있다.

이 기능은 특정 태스크가 바로 다음에 실행되지 않는 다른 태스크의 출력에 의존할 때 유용하다. 이를 위해 태스크의 context 속성을 사용한다:

# ...

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]
)

#...

​
비동기 실행

작업을 비동기적으로 실행하도록 정의할 수 있다. 이는 크루가 해당 작업이 완료될 때까지 기다리지 않고 다음 작업을 계속 진행한다는 의미다. 이 방법은 완료까지 오랜 시간이 걸리거나 다음 작업을 수행하는 데 반드시 필요하지 않은 작업에 유용하다.

나중에 실행될 작업에서 context 속성을 사용하여 비동기 작업의 출력이 완료될 때까지 기다리도록 정의할 수 있다.

#...

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, 그 역사, 그리고 흥미로운 아이디어에 관한 기사를 작성한다.",
    expected_output="AI에 관한 4단락 기사.",
    agent=writer,
    context=[list_ideas, list_important_history] # 두 작업의 출력이 완료될 때까지 기다림
)

#...

​
콜백 메커니즘

콜백 함수는 작업이 완료된 후 실행되며, 작업 결과에 따라 특정 동작을 수행하거나 알림을 보낼 수 있다.

# ...

def callback_function(output: TaskOutput):
    # 작업 완료 후 수행할 동작
    # 예: 매니저에게 이메일 전송
    print(f"""
        작업 완료!
        작업: {output.description}
        결과: {output.raw}
    """)

research_task = Task(
    description='최신 AI 뉴스 찾아 요약하기',
    expected_output='가장 중요한 AI 뉴스 5개를 요약한 목록',
    agent=research_agent,
    tools=[search_tool],
    callback=callback_function
)

#...

​
특정 작업 결과에 접근하기

크루의 실행이 완료되면, 작업 객체의 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()

# 작업 설명과 결과를 담은 TaskOutput 객체 반환
print(f"""
    작업 완료!
    작업: {task1.output.description}
    결과: {task1.output.raw}
""")

​
도구 재정의 메커니즘

작업에 도구를 지정하면 에이전트의 능력을 동적으로 조정할 수 있다. 이를 통해 CrewAI의 유연성을 극대화한다.

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

태스크를 생성하고 실행할 때, 태스크 속성의 견고성과 신뢰성을 보장하기 위해 여러 검증 메커니즘이 적용된다. 주요 검증 사항은 다음과 같다:

  • 각 태스크에 대해 하나의 출력 타입만 설정하도록 하여 출력 결과에 대한 명확한 기대를 유지한다.
  • 고유 식별자 시스템의 무결성을 유지하기 위해 id 속성을 수동으로 할당하지 못하도록 방지한다.

이러한 검증 메커니즘은 crewAI 프레임워크 내에서 태스크 실행의 일관성과 신뢰성을 유지하는 데 도움을 준다.

​
태스크 가드레일

태스크 가드레일은 한 태스크의 출력 결과를 다음 태스크로 전달하기 전에 검증, 변환, 필터링하는 강력한 방법을 제공한다. 가드레일은 선택적으로 사용하는 함수로, 다음 태스크가 시작되기 전에 실행된다. 이를 통해 태스크 출력이 특정 요구사항이나 형식을 충족하는지 확인할 수 있다.

​
기본 사용법

from typing import Tuple, Union
from crewai import Task

def validate_json_output(result: str) -> Tuple[bool, Union[dict, str]]:
    """출력이 유효한 JSON인지 검증한다."""
    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
)

​
가드레일 작동 방식

  1. 선택적 속성: 가드레일은 작업 수준에서 선택적으로 추가할 수 있는 속성이다. 필요한 곳에만 검증을 적용할 수 있다.
  2. 실행 시점: 가드레일 함수는 다음 작업이 시작되기 전에 실행된다. 이를 통해 작업 간에 유효한 데이터 흐름을 보장한다.
  3. 반환 형식: 가드레일은 반드시 (success, data) 튜플을 반환해야 한다:
    • success가 True이면 data는 검증되거나 변환된 결과다.
    • success가 False이면 data는 오류 메시지다.
  4. 결과 라우팅:
    • 성공(True) 시 결과는 자동으로 다음 작업으로 전달된다.
    • 실패(False) 시 오류는 에이전트로 전송되어 새로운 답변을 생성한다.

​
일반적인 사용 사례

​
데이터 형식 검증

def validate_email_format(result: str) -> Tuple[bool, Union[str, str]]:
    """출력값에 유효한 이메일 주소가 포함되어 있는지 확인한다."""
    import re
    email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    if re.match(email_pattern, result.strip()):
        return (True, result.strip())
    return (False, "출력값은 유효한 이메일 주소여야 한다")

​
콘텐츠 필터링

def filter_sensitive_info(result: str) -> Tuple[bool, Union[str, str]]:
    """민감한 정보를 제거하거나 검증한다."""
    sensitive_patterns = ['SSN:', 'password:', 'secret:']
    for pattern in sensitive_patterns:
        if pattern.lower() in result.lower():
            return (False, f"출력에 민감한 정보가 포함되어 있음 ({pattern})")
    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_validator(result):
        for validator in validators:
            success, data = validator(result)
            if not success:
                return (False, data)
            result = data
        return (True, result)
    return combined_validator

# 사용 예제
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  # 기본 재시도 제한을 재정의
)

​
파일 저장 시 디렉토리 생성 기능

이제 작업이 파일에 출력을 저장할 때 디렉토리를 생성할지 여부를 지정할 수 있다. 이 기능은 출력물을 체계적으로 정리하고 파일 경로를 올바르게 구성하는 데 특히 유용하다.

# ...

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
)

#...

​
결론

태스크는 CrewAI에서 에이전트의 행동을 이끄는 핵심 요소다. 태스크와 그 결과물을 올바르게 정의하면, AI 에이전트가 독립적으로 또는 협력적으로 효과적으로 작업할 수 있는 기반을 마련할 수 있다. 태스크에 적절한 도구를 제공하고, 실행 과정을 이해하며, 강력한 검증 절차를 따르는 것은 CrewAI의 잠재력을 극대화하는 데 필수적이다. 이를 통해 에이전트가 주어진 과제에 완벽히 대비하고, 태스크가 의도한 대로 실행되도록 보장할 수 있다.

AgentsCrews