조건부 태스크

조건부 태스크 (Conditional Tasks)

조건부 태스크는 이전 태스크의 결과에 따라 크루의 워크플로우를 동적으로 바꿀 수 있게 해 줍니다. 조건을 만족하지 못하면 태스크를 건너뛰고, 만족하면 실행하는 식입니다. 이렇게 하면 에이전트들이 결정을 내리고 태스크를 선택적으로 실행할 수 있어, AI 프로세스의 유연성과 효율성이 크게 올라갑니다.

출처: 공식문서

예제 사용법

from typing import List
from pydantic import BaseModel
from crewai import Agent, Crew
from crewai.tasks.conditional_task import ConditionalTask
from crewai.tasks.task_output import TaskOutput
from crewai.task import Task
from crewai_tools import SerperDevTool

# Define a condition function for the conditional task
# If false, the task will be skipped, if true, then execute the task.
def is_data_missing(output: TaskOutput) -> bool:
    return len(output.pydantic.events) < 10  # this will skip this task

# Define the agents
data_fetcher_agent = Agent(
    role="Data Fetcher",
    goal="Fetch data online using Serper tool",
    backstory="Backstory 1",
    verbose=True,
    tools=[SerperDevTool()]
)

data_processor_agent = Agent(
    role="Data Processor",
    goal="Process fetched data",
    backstory="Backstory 2",
    verbose=True
)

summary_generator_agent = Agent(
    role="Summary Generator",
    goal="Generate summary from fetched data",
    backstory="Backstory 3",
    verbose=True
)

class EventOutput(BaseModel):
    events: List[str]

task1 = Task(
    description="Fetch data about events in San Francisco using Serper tool",
    expected_output="List of 10 things to do in SF this week",
    agent=data_fetcher_agent,
    output_pydantic=EventOutput,
)

conditional_task = ConditionalTask(
    description="""
        Check if data is missing. If we have less than 10 events,
        fetch more events using Serper tool so that
        we have a total of 10 events in SF this week..
        """,
    expected_output="List of 10 Things to do in SF this week",
    condition=is_data_missing,
    agent=data_processor_agent,
)

task3 = Task(
    description="Generate summary of events in San Francisco from fetched data",
    expected_output="A complete report on the customer and their customers and competitors, including their demographics, preferences, market positioning and audience engagement.",
    agent=summary_generator_agent,
)

# Create a crew with the tasks
crew = Crew(
    agents=[data_fetcher_agent, data_processor_agent, summary_generator_agent],
    tasks=[task1, conditional_task, task3],
    verbose=True,
    planning=True
)

# Run the crew
result = crew.kickoff()
print("results", result)

핵심은 ConditionalTask와 condition 함수입니다. condition은 TaskOutput을 받아 bool을 반환하는 함수로, False를 반환하면 해당 태스크가 건너뛰어지고 True면 실행됩니다. 위 예시에서는 output.pydantic.events의 개수가 10보다 작으면 데이터가 부족하다고 판단해 추가 데이터 수집 태스크를 실행하도록 했습니다.

더 알아보기