태스크: 에이전트가 수행할 작업 단위
태스크: 에이전트가 수행할 작업 단위
CrewAI에서 태스크(Task)는 에이전트가 수행하는 구체적인 작업 단위예요. 설명·담당 에이전트·필요한 툴 등 실행에 필요한 모든 정보를 담아 다양한 복잡도의 작업을 처리하게 합니다. 태스크는 여러 에이전트가 협력하도록 만들 수도 있고, 크루의 프로세스가 순서·할당을 조율합니다. 여기서는 태스크 생성부터 출력, 의존성·컨텍스트, 가드레일, 구조화 출력까지 다룰게요.
출처: 공식문서
본문
개요
CrewAI 프레임워크에서 Task는 Agent가 완료하는 특정 임무입니다. 태스크는 설명, 담당 에이전트, 필요한 툴 등 실행에 필요한 모든 세부 정보를 제공해 다양한 복잡도의 액션을 지원합니다.
CrewAI 내 태스크는 협력적일 수 있어 여러 에이전트가 함께 작업해야 합니다. 이는 태스크 속성으로 관리되고 크루의 프로세스가 오케스트레이션하여 팀워크와 효율성을 높입니다.
태스크 실행 흐름
태스크는 두 가지 방식으로 실행됩니다:
- Sequential(순차): 태스크가 정의된 순서대로 실행
- Hierarchical(계층): 태스크가 에이전트의 역할·전문성에 따라 할당
실행 흐름은 크루 생성 시 정의합니다:
crew = Crew(
agents=[agent1, agent2],
tasks=[task1, task2],
process=Process.sequential # or Process.hierarchical
)
태스크 속성
| 속성 | 파라미터 | 타입 | 설명 |
|---|---|---|---|
| Description | description |
str |
태스크가 무엇인지에 대한 명확하고 간결한 진술 |
| Expected Output | expected_output |
str |
태스크 완료가 어떤 모습인지에 대한 상세 설명 |
| Name (선택) | name |
Optional[str] |
태스크 이름 식별자 |
| Agent (선택) | agent |
Optional[BaseAgent] |
태스크 실행을 담당하는 에이전트 |
| Tools (선택) | tools |
List[BaseTool] |
이 태스크에서 에이전트가 사용하도록 제한된 툴 |
| Context (선택) | context |
Optional[List["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]] |
태스크별 설정 파라미터 |
| Output File (선택) | output_file |
Optional[str] |
태스크 출력을 저장할 파일 경로 |
| 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]] |
태스크 출력용 Pydantic 모델 |
| Callback (선택) | callback |
Optional[Any] |
태스크 완료 후 실행할 함수/객체 |
| Guardrail (선택) | guardrail |
Optional[Callable] |
다음 태스크 진행 전 태스크 출력을 검증하는 함수 |
| Guardrails (선택) | guardrails |
Optional[List[Callable]] |
다음 태스크 진행 전 출력을 검증하는 가드레일 목록 |
| Guardrail Max Retries (선택) | guardrail_max_retries |
Optional[int] |
가드레일 검증 실패 시 최대 재시도 횟수. 기본값 3 |
태스크 생성
태스크를 만드는 두 가지 일반적인 방법은 JSONC 프로젝트 구성(신규 크루 권장) 과 코드에서 직접 정의입니다.
JSONC 구성(권장)
crewai create crew <name>으로 만든 새 프로젝트는 crew.jsonc에 태스크를 정의합니다. agents 배열은 agents/의 파일을 가리키고, tasks 배열은 크루가 실행할 순서 있는 작업을 정의합니다.
{placeholder} 값을 태스크 description, expected_output, output_file에 사용할 수 있고 기본값은 최상위 inputs 객체에 둡니다. crewai run은 누락된 값을 묻습니다.
두 개의 순서 태스크가 있는 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"
}
}
각 태스크는 description과 expected_output을 포함해야 합니다. agent 값은 agents에 나열된 에이전트 이름과 일치해야 합니다. context는 이전 태스크 이름 목록이며 순방향 참조는 거부되어 순차 컨텍스트가 명시적으로 유지됩니다.
태스크 항목은 공개 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. 조건 태스크는 condition 필드와 함께 "type": "ConditionalTask"를 사용합니다.
코드 직접 정의(대안)
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, # Enable markdown formatting for the final output
output_file="report.md"
)
태스크 출력
CrewAI 프레임워크에서 태스크 출력은 TaskOutput 클래스에 캡슐화됩니다. raw 출력, JSON, Pydantic 모델 등 다양한 형식으로 태스크 결과에 접근하는 구조적 방법을 제공합니다.
기본적으로 TaskOutput은 raw 출력만 포함합니다. TaskOutput은 원본 Task 객체가 각각 output_pydantic 또는 output_json으로 구성된 경우에만 pydantic 또는 json_dict 출력을 포함합니다.
태스크 출력 속성
| 속성 | 파라미터 | 타입 | 설명 |
|---|---|---|---|
| Description | description |
str |
태스크 설명 |
| Summary | summary |
Optional[str] |
태스크 요약. 설명의 처음 10단어에서 자동 생성 |
| Raw | raw |
str |
태스크의 원본 출력. 기본 출력 형식 |
| Pydantic | pydantic |
Optional[BaseModel] |
태스크의 구조화 출력을 나타내는 Pydantic 모델 객체 |
| JSON Dict | json_dict |
Optional[Dict[str, Any]] |
태스크의 JSON 출력을 나타내는 딕셔너리 |
| Agent | agent |
str |
태스크를 실행한 에이전트 |
| Output Format | output_format |
OutputFormat |
태스크 출력 형식. RAW, JSON, Pydantic 옵션. 기본 RAW |
| Messages | messages |
list[LLMMessage] |
마지막 태스크 실행의 메시지 |
태스크 출력 접근
# Example 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]
)
# Execute the crew
crew = Crew(
agents=[research_agent],
tasks=[task],
verbose=True
)
result = crew.kickoff()
# Accessing the task output
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 파라미터는 태스크 출력의 자동 Markdown 포맷팅을 활성화합니다. True로 설정하면 태스크가 에이전트에게 적절한 Markdown 문법으로 최종 답을 포맷하도록 지시합니다. 에이전트는 #(제목), **text**(굵게), *text*(기울임), -·*(불릿), `code`(인라인 코드), ```language(코드 블록)을 받습니다.
YAML 구성에서도 활성화할 수 있습니다:
analysis_task:
description: >
Analyze the market data and create a detailed report
expected_output: >
A comprehensive analysis with charts and key findings
agent: analyst
markdown: true # Enable markdown formatting
output_file: analysis.md
태스크 의존성과 컨텍스트
태스크는 context 속성으로 다른 태스크의 출력에 의존할 수 있습니다:
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] # This task will wait for research_task to complete
)
태스크 가드레일
태스크 가드레일은 출력이 다음 태스크로 전달되기 전 검증·변환하는 방법을 제공합니다. 두 가지 유형을 지원합니다:
- 함수 기반 가드레일: 사용자 정의 검증 로직을 가진 파이썬 함수. 검증 과정을 완전히 제어하고 신뢰할 수 있는 결정적 결과를 보장합니다.
- LLM 기반 가드레일: 에이전트의 LLM으로 자연어 기준에 따라 출력을 검증하는 문자열 설명. 복잡하거나 주관적인 검증 요구에 적합합니다.
함수 기반 가드레일
guardrail 파라미터로 검증 함수를 제공합니다:
from typing import Tuple, Union, Dict, Any
from crewai import TaskOutput
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
"""Validate blog content meets requirements."""
try:
# Check word count
word_count = len(result.raw.split())
if word_count > 200:
return (False, "Blog content exceeds 200 words")
# Additional validation logic here
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 # Add the guardrail function
)
LLM 기반 가드레일(문자열 설명)
guardrail 또는 guardrails 파라미터에 문자열을 제공하면 CrewAI가 에이전트의 LLM으로 출력을 검증하는 LLMGuardrail을 자동 생성합니다. 요구사항: 태스크에 agent가 지정되어야 하고(가드레일이 에이전트의 LLM을 사용), 검증 기준을 설명하는 명확한 문자열을 제공해야 합니다.
from crewai import Task
# Single LLM-based guardrail
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 가드레일은 출력을 기준과 대조해 분석하고, 준수하면 (True, output), 실패하면 특정 피드백과 함께 (False, feedback)를 반환합니다.
여러 가드레일
guardrails 파라미터로 여러 가드레일을 적용할 수 있습니다. 순차적으로 실행되며 각 가드레일은 이전 가드레일의 출력을 받습니다. 참고: guardrails가 제공되면 guardrail보다 우선하며, guardrails가 설정되면 guardrail 파라미터는 무시됩니다.
가드레일이 (False, error)를 반환하면 오류가 에이전트로 전송되고, 에이전트가 문제를 고치려 시도하며 guardrail_max_retries에 도달할 때까지 반복합니다.
가드레일 함수 요구사항
- 함수 시그니처: 정확히 하나의 파라미터(태스크 출력)를 받아야 하고
(bool, Any)튜플을 반환해야 합니다. 타입 힌트는 권장되지만 선택 사항입니다. - 반환 값: 성공 시
(True, validated_result), 실패 시(False, "Error message explain the failure").
태스크에서 구조화·일관된 출력 얻기
output_pydantic 사용
output_pydantic 속성으로 태스크 출력이 따라야 할 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="Generate a blog title and content",
backstory="""You are an expert content creator, skilled in crafting engaging and informative blog posts.""",
verbose=False,
allow_delegation=False,
llm="gpt-4o",
)
task1 = Task(
description="""Create a blog title and content on a given topic. Make sure the content is under 200 words.""",
expected_output="A compelling blog title and well-written content.",
agent=blog_agent,
output_pydantic=Blog,
)
# Instantiate your crew with a sequential process
crew = Crew(
agents=[blog_agent],
tasks=[task1],
verbose=True,
process=Process.sequential,
)
result = crew.kickoff()
# Option 1: Accessing Properties Using Dictionary-Style Indexing
print("Accessing Properties - Option 1")
title = result["title"]
content = result["content"]
print("Title:", title)
print("Content:", content)
# Option 2: Accessing Properties Directly from the Pydantic Model
print("Accessing Properties - Option 2")
title = result.pydantic.title
content = result.pydantic.content
print("Title:", title)
print("Content:", content)
# Option 3: Accessing Properties Using the to_dict() Method
print("Accessing Properties - Option 3")
output_dict = result.to_dict()
title = output_dict["title"]
content = output_dict["content"]
print("Title:", title)
print("Content:", content)
출력 접근 방식: ①딕셔너리 스타일 인덱싱(result["field_name"] — CrewOutput이 __getitem__ 구현), ②result.pydantic에서 직접 접근, ③to_dict()로 딕셔너리 변환, ④객체 전체 출력.
output_json 사용
output_json 속성으로 JSON 형식의 예상 출력을 정의할 수 있습니다. 태스크 출력이 애플리케이션에서 쉽게 파싱·사용할 수 있는 유효한 JSON 구조임을 보장합니다. output_pydantic과 동일하게 Blog 모델을 output_json=Blog로 지정하고, 딕셔너리 스타일 인덱싱 또는 객체 출력으로 접근합니다.
태스크에 툴 통합
CrewAI Toolkit·LangChain Tools의 툴을 활용해 태스크 성능과 에이전트 상호작용을 향상할 수 있습니다.
import os
os.environ["OPENAI_API_KEY"] = "Your Key"
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key
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
)
# to perform a semantic search for a specified query from a text's content across the internet
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에서 한 태스크의 출력은 다음 태스크로 자동 전달되지만, 다른 태스크(여러 개 포함)의 출력을 컨텍스트로 명시적으로 지정할 수도 있습니다. async_execution=True 태스크는 다음 태스크가 완료를 기다리지 않게 하며, 이후 태스크에서 context로 기다릴 출력을 지정할 수 있습니다.
콜백 메커니즘
콜백 함수는 태스크 완료 후 실행되어 태스크 결과에 기반한 액션·알림을 트리거합니다.
def callback_function(output: TaskOutput):
# Do something after the task is completed
# Example: Send an email to the manager
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
)
특정 태스크 출력 접근
크루가 끝난 뒤 태스크 객체의 output 속성으로 특정 태스크의 출력에 접근할 수 있습니다: task1.output.description, task1.output.raw 등.
파일 저장 시 디렉터리 생성
create_directory 파라미터는 태스크 출력을 파일로 저장할 때 CrewAI가 디렉터리를 자동 생성할지 제어합니다.
- 기본 동작:
create_directory=True(기본값) — 출력 파일 경로에서 누락된 디렉터리 자동 생성. - 비활성화:
create_directory=False— 디렉터리가 이미 존재해야 합니다. 존재하지 않으면RuntimeError발생.
오류 처리·검증 메커니즘
- 태스크당 하나의 출력 유형만 설정하도록 보장해 출력 기대치를 명확히 유지.
id속성의 수동 할당을 방지해 고유 식별자 시스템의 무결성 보장.
결론
태스크는 CrewAI에서 에이전트 행동을 이끄는 원동력입니다. 태스크와 결과물을 제대로 정의하면 AI 에이전트가 독립적으로든 협력 단위로든 효과적으로 작업할 수 있습니다. 적절한 툴 장착, 실행 프로세스 이해, 견고한 검증 실천이 CrewAI의 잠재력을 최대화하는 핵심입니다.