크루 비동기 실행

크루 비동기 실행 (Kickoff Crew Asynchronously)

CrewAI는 크루를 비동기로 실행해 실행 체인을 블로킹하지 않고 시작하게 해 줍니다. 여러 크루를 동시에 돌리거나, 크루가 실행되는 동안 다른 작업을 해야 할 때 특히 유용합니다. CrewAI는 네이티브 async 방식과 스레드 기반 방식 두 가지를 제공하며, 상황에 맞게 고르면 됩니다.

출처: 공식문서

개요

높은 동시성 워크로드에는 akickoff()가 권장됩니다. 태스크 실행, 메모리 연산, 지식 검색 전반에 네이티브 async를 사용하기 때문입니다.

akickoff()로 네이티브 async 실행

akickoff() 메서드는 진정한 네이티브 async 실행을 제공하며, 태스크 실행·메모리 연산·지식 조회를 포함한 전체 실행 체인에서 async/await를 사용합니다.

메서드 시그니처

async def akickoff(self, inputs: dict) -> CrewOutput:

파라미터

• inputs (dict): 태스크에 필요한 입력 데이터를 담은 딕셔너리입니다.

반환값

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

예제: 네이티브 async 크루 실행

import asyncio
from crewai import Crew, Agent, Task

# Create an agent
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
)

# Create a task
data_analysis_task = Task(
    description="Analyze the given dataset and calculate the average age of participants. Ages: {ages}",
    agent=coding_agent,
    expected_output="The average age of the participants."
)

# Create a crew
analysis_crew = Crew(
    agents=[coding_agent],
    tasks=[data_analysis_task]
)

# Native async execution
async def main():
    result = await analysis_crew.akickoff(inputs={"ages": [25, 30, 35, 40, 45]})
    print("Crew Result:", result)

asyncio.run(main())

예제: 여러 네이티브 async 크루

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. Ages: {ages}",
    agent=coding_agent,
    expected_output="The average age of the participants."
)

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

crew_1 = Crew(agents=[coding_agent], tasks=[task_1])
crew_2 = Crew(agents=[coding_agent], tasks=[task_2])

async def main():
    results = await asyncio.gather(
        crew_1.akickoff(inputs={"ages": [25, 30, 35, 40, 45]}),
        crew_2.akickoff(inputs={"ages": [20, 22, 24, 28, 30]})
    )

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

asyncio.run(main())

네이티브 async로 여러 크루를 동시에 실행하려면 asyncio.gather()를 사용합니다.

예제: 여러 입력에 대한 네이티브 async

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 dataset and calculate the average age. Ages: {ages}",
    agent=coding_agent,
    expected_output="The average age of the participants."
)

analysis_crew = Crew(
    agents=[coding_agent],
    tasks=[data_analysis_task]
)

async def main():
    datasets = [
        {"ages": [25, 30, 35, 40, 45]},
        {"ages": [20, 22, 24, 28, 30]},
        {"ages": [30, 35, 40, 45, 50]}
    ]

    results = await analysis_crew.akickoff_for_each(datasets)

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

asyncio.run(main())

akickoff_for_each()를 쓰면 여러 입력에 대해 네이티브 async로 크루를 동시에 실행할 수 있습니다.

kickoff_async()로 스레드 기반 async

kickoff_async() 메서드는 동기 kickoff()를 스레드로 감싸 async 실행을 제공합니다. 더 간단한 async 통합이나 하위 호환성(backward compatibility)이 필요할 때 유용합니다.

메서드 시그니처

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

파라미터

• inputs (dict): 태스크에 필요한 입력 데이터를 담은 딕셔너리입니다.

반환값

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

예제: 스레드 기반 async 실행

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,
    expected_output="The average age of the participants."
)

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

예제: 여러 스레드 기반 async 크루

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,
    expected_output="The average age of the participants."
)

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

crew_1 = Crew(agents=[coding_agent], tasks=[task_1])
crew_2 = Crew(agents=[coding_agent], tasks=[task_2])

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

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

비동기 스트리밍 (Async Streaming)

import asyncio
from crewai import Crew, Agent, Task

agent = Agent(
    role="Researcher",
    goal="Research and summarize topics",
    backstory="You are an expert researcher."
)

task = Task(
    description="Research the topic: {topic}",
    agent=agent,
    expected_output="A comprehensive summary of the topic."
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    stream=True  # Enable streaming
)

async def main():
    streaming_output = await crew.akickoff(inputs={"topic": "AI trends in 2024"})

    # Async iteration over streaming chunks
    async for chunk in streaming_output:
        print(f"Chunk: {chunk.content}")

    # Access final result after streaming completes
    result = streaming_output.result
    print(f"Final result: {result.raw}")

asyncio.run(main())

크루에 stream=True가 설정되면 두 async 메서드 모두 스트리밍을 지원합니다.

주요 사용 사례

• 병렬 콘텐츠 생성 (Parallel Content Generation): 서로 다른 주제의 콘텐츠를 생성하는 여러 독립 크루를 비동기로 시작합니다. 예를 들어 한 크루는 AI 트렌드 기사를 조사·초안 작성하고, 다른 크루는 신제품 출시에 관한 소셜 미디어 포스트를 만듭니다.
• 동시 시장 조사 (Concurrent Market Research Tasks): 여러 크루를 비동기로 실행해 시장 조사를 병렬로 진행합니다. 한 크루는 산업 트렌드를, 다른 크루는 경쟁사 전략을, 또 다른 크루는 소비자 감정을 분석합니다.
• 독립 여행 계획 모듈 (Independent Travel Planning Modules): 여행의 서로 다른 측면을 독립적으로 계획하는 별도 크루를 실행합니다. 한 크루는 항공편, 다른 크루는 숙소, 세 번째 크루는 액티비티를 처리합니다.

akickoff() vs kickoff_async() 선택하기

더 알아보기

• Kickoff Crew for Each — 리스트의 각 항목마다 크루 실행
• Streaming 개념
• CrewAI 공식 개념: Crews