CrewAI 프레임워크에서 Task는 Agent가 완료하는 구체적인 과제입니다.Task는 실행을 위한 모든 세부 정보를 제공합니다. 여기에는 설명, 책임 Agent, 필요한 도구 등이 포함되어 다양한 작업 복잡성을 지원합니다.CrewAI 내의 Task는 협업이 가능하며, 여러 Agent가 함께 작업해야 할 수도 있습니다. 이는 Task 속성을 통해 관리되며, Crew의 프로세스를 통해 조율되어 팀워크와 효율성을 향상시킵니다.
CrewAI 엔터프라이즈에는 복잡한 Task 생성과 연결을 단순화하는 Crew Studio의 비주얼 Task 빌더가 포함되어 있습니다. Task 흐름을 시각적으로 설계하고, 코드를 작성하지 않고도 실시간으로 테스트해 볼 수 있습니다.비주얼 Task 빌더의 주요 기능:
YAML 구성을 사용하면 작업을 정의할 때 더 깔끔하고 유지 관리가 용이한 방법을 제공합니다. CrewAI 프로젝트에서 작업을 정의할 때 이 방식을 사용하는 것을 강력히 권장합니다.설치 섹션에 따라 CrewAI 프로젝트를 생성한 후, src/latest_ai_development/config/tasks.yaml 파일로 이동하여 템플릿을 귀하의 특정 작업 요구 사항에 맞게 수정하세요.
YAML 파일 내 변수(예: {topic})는 크루를 실행할 때 입력값에서 가져온 값으로 대체됩니다:
Code
Copy
Ask AI
crew.kickoff(inputs={'topic': 'AI Agents'})
아래는 YAML을 사용하여 작업을 구성하는 방법의 예시입니다:
tasks.yaml
Copy
Ask AI
research_task: description: > Conduct a thorough research about {topic} 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 {topic} agent: researcherreporting_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. Formatted as markdown without '```' agent: reporting_analyst markdown: true output_file: report.md
이 YAML 구성을 코드에서 사용하려면 CrewBase를 상속받는 크루 클래스를 생성하세요:
from crewai import Taskresearch_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")
agent를 직접 지정하여 할당하거나, 역할·가용성 등에 따라 hierarchical CrewAI 프로세스가 자동으로 결정하도록 둘 수 있습니다.
작업 결과를 이해하는 것은 효과적인 AI 워크플로우를 구축하는 데 매우 중요합니다. CrewAI는 여러 출력 형식을 지원하는 TaskOutput 클래스를 통해 작업 결과를 구조적으로 처리할 수 있는 방식을 제공합니다. 이 클래스는 작업 간에 출력값을 쉽게 전달할 수 있도록 지원합니다.CrewAI 프레임워크에서 작업의 출력은 TaskOutput 클래스에 캡슐화되어 있습니다. 이 클래스는 작업의 결과를 구조적으로 접근할 수 있도록 해주며, raw 출력, JSON, Pydantic 모델 등 다양한 형식을 지원합니다.기본적으로 TaskOutput에는 raw 출력만 포함됩니다. 원래의 Task 객체가 각각 output_pydantic 또는 output_json으로 구성된 경우에만 TaskOutput에 pydantic 또는 json_dict 출력이 포함됩니다.
# Example tasktask = 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 crewcrew = Crew( agents=[research_agent], tasks=[task], verbose=True)result = crew.kickoff()# Accessing the task outputtask_output = task.outputprint(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}")
# Example task with markdown formatting enabledformatted_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 # Enable automatic markdown formatting)
markdown=True일 때, 에이전트는 다음과 같이 출력을 포매팅하라는 추가 지시를 받게 됩니다:
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 속성을 사용하여 다른 작업의 출력에 의존할 수 있습니다. 예를 들어:
Code
Copy
Ask AI
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)
작업 가드레일은 작업 출력물을 다음 작업에 전달하기 전에 유효성을 검사하고 변환할 수 있는 방식을 제공합니다. 이 기능은 데이터 품질을 보장하고 에이전트의 출력이 특정 기준을 충족하지 않을 때 피드백을 제공하는 데 도움이 됩니다.가드레일은 사용자 지정 유효성 검사 로직을 포함하는 Python 함수로 구현되며, 유효성 검사 프로세스를 완전히 제어할 수 있어 신뢰할 수 있고 결정적인 결과를 보장합니다.
output_pydantic 속성을 사용하면 작업 출력이 준수해야 할 Pydantic 모델을 정의할 수 있습니다. 이를 통해 출력이 구조화될 뿐만 아니라 Pydantic 모델에 따라 유효성 검증도 보장할 수 있습니다.다음은 output_pydantic을 사용하는 방법을 보여주는 예제입니다.
Code
Copy
Ask AI
import jsonfrom crewai import Agent, Crew, Process, Taskfrom pydantic import BaseModelclass Blog(BaseModel): title: str content: strblog_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 processcrew = Crew( agents=[blog_agent], tasks=[task1], verbose=True, process=Process.sequential,)result = crew.kickoff()# Option 1: Accessing Properties Using Dictionary-Style Indexingprint("Accessing Properties - Option 1")title = result["title"]content = result["content"]print("Title:", title)print("Content:", content)# Option 2: Accessing Properties Directly from the Pydantic Modelprint("Accessing Properties - Option 2")title = result.pydantic.titlecontent = result.pydantic.contentprint("Title:", title)print("Content:", content)# Option 3: Accessing Properties Using the to_dict() Methodprint("Accessing Properties - Option 3")output_dict = result.to_dict()title = output_dict["title"]content = output_dict["content"]print("Title:", title)print("Content:", content)# Option 4: Printing the Entire Blog Objectprint("Accessing Properties - Option 5")print("Blog:", result)
이 예제에서:
title과 content 필드를 가진 Pydantic 모델 Blog가 정의되어 있습니다.
작업 task1은 output_pydantic 속성을 사용하여 출력이 Blog 모델을 준수해야 함을 명시합니다.
output_json 속성을 사용하면 예상되는 출력을 JSON 형식으로 정의할 수 있습니다. 이를 통해 태스크의 출력이 쉽게 파싱되고, 애플리케이션에서 사용할 수 있는 유효한 JSON 구조임을 보장합니다.다음은 output_json 사용 방법을 보여주는 예시입니다:
Code
Copy
Ask AI
import jsonfrom crewai import Agent, Crew, Process, Taskfrom pydantic import BaseModel# Define the Pydantic model for the blogclass Blog(BaseModel): title: str content: str# Define the agentblog_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",)# Define the task with output_json set to the Blog modeltask1 = Task( description="""Create a blog title and content on a given topic. Make sure the content is under 200 words.""", expected_output="A JSON object with 'title' and 'content' fields.", agent=blog_agent, output_json=Blog,)# Instantiate the crew with a sequential processcrew = Crew( agents=[blog_agent], tasks=[task1], verbose=True, process=Process.sequential,)# Kickoff the crew to execute the taskresult = crew.kickoff()# Option 1: Accessing Properties Using Dictionary-Style Indexingprint("Accessing Properties - Option 1")title = result["title"]content = result["content"]print("Title:", title)print("Content:", content)# Option 2: Printing the Entire Blog Objectprint("Accessing Properties - Option 2")print("Blog:", result)
이 예시에서:
Pydantic 모델인 Blog가 title과 content 필드로 정의되어 있으며, 이는 JSON 출력의 구조를 명시하는 데 사용됩니다.
태스크 task1은 output_json 속성을 사용하여 Blog 모델에 부합하는 JSON 출력을 기대함을 나타냅니다.
딕셔너리 스타일 인덱싱을 사용하여 속성 접근하기: result[“field_name”]과 같이 필드를 직접 접근할 수 있습니다. 이는 CrewOutput 클래스가 getitem 메서드를 구현하고 있어 출력을 딕셔너리처럼 사용할 수 있기 때문입니다. 이 방법에서는 result에서 title과 content를 가져옵니다.
전체 블로그 객체 출력하기: result를 출력하면 CrewOutput 객체의 문자열 표현을 얻을 수 있습니다. str 메서드가 JSON 출력을 반환하도록 구현되어 있기 때문에, 전체 출력을 Blog 객체를 나타내는 형식이 잘 갖추어진 문자열로 볼 수 있습니다.
output_pydantic 또는 output_json을 사용하면, 작업의 출력이 일관되고 구조화된 형식으로 생성되므로 애플리케이션 내 또는 여러 작업 간에 데이터를 더 쉽게 처리하고 활용할 수 있습니다.
import osos.environ["OPENAI_API_KEY"] = "Your Key"os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API keyfrom crewai import Agent, Task, Crewfrom crewai_tools import SerperDevToolresearch_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 internetsearch_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)
이 예시는 특정 도구와 함께 사용되는 task가 맞춤형 task 실행을 위해 에이전트의 기본 도구 세트를 어떻게 재정의할 수 있는지 보여줍니다.
CrewAI에서는 한 작업의 출력이 자동으로 다음 작업으로 전달되지만, 특정 작업(여러 개 포함)의 출력을 다른 작업의 컨텍스트로 명확하게 지정할 수도 있습니다.이는 한 작업이 바로 뒤에 수행되지 않는 다른 작업의 출력에 의존해야 할 때 유용합니다. 이는 작업의 context 속성을 통해 수행됩니다:
Code
Copy
Ask AI
# ...research_ai_task = Task( description="Research the latest developments in AI", expected_output="A list of recent AI developments", async_execution=True, agent=research_agent, tools=[search_tool])research_ops_task = Task( description="Research the latest developments in AI Ops", expected_output="A list of recent AI Ops developments", async_execution=True, agent=research_agent, tools=[search_tool])write_blog_task = Task( description="Write a full blog post about the importance of AI and its latest news", expected_output="Full blog post that is 4 paragraphs long", agent=writer_agent, context=[research_ai_task, research_ops_task])#...
작업을 비동기로 실행되도록 정의할 수 있습니다. 이는 crew가 해당 작업이 완료될 때까지 기다리지 않고 다음 작업을 계속 진행한다는 것을 의미합니다. 시간이 오래 걸리는 작업이거나, 이후 작업 수행에 필수적이지 않은 작업에 유용합니다.이후 작업에서 비동기 작업의 출력이 완료될 때까지 기다리도록 하려면, context 속성을 사용할 수 있습니다.
Code
Copy
Ask AI
#...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 # Will be executed asynchronously)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 # Will be executed asynchronously)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] # Will wait for the output of the two tasks to be completed)#...
콜백 함수는 작업이 완료된 후 실행되며, 작업 결과에 따라 동작 또는 알림을 트리거할 수 있습니다.
Code
Copy
Ask AI
# ...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)#...
crew가 실행을 마치면, 해당 task 객체의 output 속성을 사용하여 특정 task의 output에 접근할 수 있습니다:
Code
Copy
Ask AI
# ...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()# Returns a TaskOutput object with the description and results of the taskprint(f""" Task completed! Task: {task1.output.description} Output: {task1.output.raw}""")
create_directory 매개변수는 CrewAI가 작업 결과를 파일로 저장할 때 디렉토리를 자동으로 생성할지 여부를 제어합니다. 이 기능은 출력물을 체계적으로 정리하고, 특히 복잡한 프로젝트 계층 구조에서 파일 경로가 올바르게 구조화되도록 보장하는 데 매우 유용합니다.
기본적으로 create_directory=True로 설정되어 있으며, 이는 CrewAI가 출력 파일 경로에 누락된 디렉토리를 자동으로 생성함을 의미합니다:
Code
Copy
Ask AI
# 기본 동작 - 디렉토리가 자동으로 생성됩니다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)
작업(task)은 CrewAI 에이전트의 행동을 이끄는 원동력입니다.
작업과 그 결과를 적절하게 정의함으로써, 에이전트가 독립적으로 또는 협업 단위로 효과적으로 작동할 수 있는 기반을 마련할 수 있습니다.
작업에 적합한 도구를 장착하고, 실행 과정을 이해하며, 견고한 검증 절차를 따르는 것은 CrewAI의 잠재력을 극대화하는 데 필수적입니다.
이를 통해 에이전트가 할당된 작업에 효과적으로 준비되고, 작업이 의도대로 수행될 수 있습니다.