Skip to main content

​
소개

CrewAI는 크루를 비동기적으로 실행할 수 있는 기능을 제공한다. 이를 통해 크루 실행을 논블로킹 방식으로 시작할 수 있다. 이 기능은 여러 크루를 동시에 실행하거나 크루가 실행되는 동안 다른 작업을 수행해야 할 때 특히 유용하다.

​
비동기 크루 실행

크루를 비동기적으로 실행하려면 kickoff_async() 메서드를 사용한다. 이 메서드는 크루 실행을 별도의 스레드에서 시작하며, 메인 스레드는 다른 작업을 계속 실행할 수 있게 한다.

​
메서드 시그니처

def kickoff_async(self, inputs: dict) -> CrewOutput:

​
파라미터

  • inputs (dict): 작업에 필요한 입력 데이터를 담은 딕셔너리

​
반환값

  • CrewOutput: 크루 실행 결과를 나타내는 객체

​
잠재적 사용 사례

  • 병렬 콘텐츠 생성: 여러 독립적인 크루를 비동기적으로 실행해 각기 다른 주제의 콘텐츠를 생성한다. 예를 들어, 한 크루는 AI 트렌드에 관한 기사를 조사하고 작성하는 동안, 다른 크루는 신제품 출시와 관련된 소셜 미디어 게시물을 만든다. 각 크루는 독립적으로 작동해 콘텐츠 생산을 효율적으로 확장할 수 있다.
  • 동시적 시장 조사 작업: 여러 크루를 비동기적으로 실행해 시장 조사를 병렬로 수행한다. 한 크루는 산업 동향을 분석하고, 다른 크루는 경쟁사의 전략을 조사하며, 또 다른 크루는 소비자 감정을 평가한다. 각 크루는 독립적으로 작업을 완료해 더 빠르고 포괄적인 통찰력을 얻을 수 있다.
  • 독립적인 여행 계획 모듈: 여행의 다양한 측면을 독립적으로 계획하기 위해 별도의 크루를 실행한다. 한 크루는 항공편 옵션을 처리하고, 다른 크루는 숙박 시설을 담당하며, 세 번째 크루는 활동을 계획한다. 각 크루는 비동기적으로 작동해 여행의 다양한 구성 요소를 동시에 독립적으로 계획할 수 있어 더 빠른 결과를 얻을 수 있다.

​
예제: 단일 비동기 크루 실행

다음은 asyncio를 사용해 크루를 비동기적으로 실행하고 결과를 기다리는 예제다: 
import asyncio
from crewai import Crew, Agent, Task

# 코드 실행이 가능한 에이전트 생성
coding_agent = Agent(
    role="Python Data Analyst",
    goal="Analyze data and provide insights using Python",
    backstory="You are an experienced data analyst with strong Python skills.",
    allow_code_execution=True
)

# 코드 실행이 필요한 태스크 생성
data_analysis_task = Task(
    description="Analyze the given dataset and calculate the average age of participants. Ages: {ages}",
    agent=coding_agent
)

# 크루 생성 및 태스크 추가
analysis_crew = Crew(
    agents=[coding_agent],
    tasks=[data_analysis_task]
)

# 크루를 비동기적으로 실행하는 함수
async def async_crew_execution():
    result = await analysis_crew.kickoff_async(inputs={"ages": [25, 30, 35, 40, 45]})
    print("Crew Result:", result)

# 비동기 함수 실행
asyncio.run(async_crew_execution())

​
예제: 다중 비동기 Crew 실행

이 예제에서는 asyncio.gather()를 사용해 여러 개의 crew를 비동기적으로 실행하고, 모든 작업이 완료될 때까지 기다리는 방법을 보여준다: 
import asyncio
from crewai import Crew, Agent, Task

# 코드 실행이 허용된 에이전트 생성
coding_agent = Agent(
    role="Python Data Analyst",
    goal="Analyze data and provide insights using Python",
    backstory="You are an experienced data analyst with strong Python skills.",
    allow_code_execution=True
)

# 코드 실행이 필요한 태스크 생성
task_1 = Task(
    description="Analyze the first dataset and calculate the average age of participants. Ages: {ages}",
    agent=coding_agent
)

task_2 = Task(
    description="Analyze the second dataset and calculate the average age of participants. Ages: {ages}",
    agent=coding_agent
)

# 두 개의 crew를 생성하고 태스크 추가
crew_1 = Crew(agents=[coding_agent], tasks=[task_1])
crew_2 = Crew(agents=[coding_agent], tasks=[task_2])

# 여러 crew를 비동기적으로 실행하고 모든 작업이 완료될 때까지 기다리는 비동기 함수
async def async_multiple_crews():
    result_1 = crew_1.kickoff_async(inputs={"ages": [25, 30, 35, 40, 45]})
    result_2 = crew_2.kickoff_async(inputs={"ages": [20, 22, 24, 28, 30]})

    # 두 crew가 모두 완료될 때까지 기다림
    results = await asyncio.gather(result_1, result_2)

    for i, result in enumerate(results, 1):
        print(f"Crew {i} Result:", result)

# 비동기 함수 실행
asyncio.run(async_multiple_crews())
⌘I