CrewAI Run Automation 도구

CrewAI Run Automation 도구 (InvokeCrewAIAutomationTool)

CrewAI의 InvokeCrewAIAutomationTool은 CrewAI Platform API와 외부 크루 서비스를 연동하는 도구예요. CrewAI 에이전트 안에서 CrewAI Platform 자동화를 호출하고 상호작용할 수 있게 해줘서, 서로 다른 크루 워크플로 사이를 매끄럽게 연결해 준답니다.

출처: 문서

본문

InvokeCrewAIAutomationTool은 CrewAI Platform API를 외부 크루 서비스와 통합해줍니다. 이 도구를 사용하면 CrewAI 에이전트 안에서 CrewAI Platform 자동화를 호출하고 상호작용할 수 있어, 서로 다른 크루 워크플로 간의 원활한 통합이 가능해요.

설치 (Installation)

uv pip install 'crewai[tools]'

사전 요구사항 (Requirements)

CrewAI Platform API 접근 권한
인증을 위한 유효한 bearer token
CrewAI Platform automation 엔드포인트에 대한 네트워크 접근

사용법 (Usage)

CrewAI 에이전트와 함께 이 도구를 사용하는 방법은 다음과 같습니다.

from crewai import Agent, Task, Crew
from crewai_tools import InvokeCrewAIAutomationTool

# 도구 초기화
automation_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://data-analysis-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Data Analysis Crew",
    crew_description="Analyzes data and generates insights"
)

# 이 도구를 사용하는 CrewAI 에이전트 생성
automation_coordinator = Agent(
    role='Automation Coordinator',
    goal='Coordinate and execute automated crew tasks',
    backstory='I am an expert at leveraging automation tools to execute complex workflows.',
    tools=[automation_tool],
    verbose=True
)

# 에이전트를 위한 태스크 생성
analysis_task = Task(
    description="Execute data analysis automation and provide insights",
    agent=automation_coordinator,
    expected_output="Comprehensive data analysis report"
)

# 에이전트로 크루 생성
crew = Crew(
    agents=[automation_coordinator],
    tasks=[analysis_task],
    verbose=2
)

# 크루 실행
result = crew.kickoff()
print(result)

도구 인자 (Tool Arguments)

Argument	Type	Required	Default	Description
crew_api_url	`str`	Yes	None	CrewAI Platform 자동화 API의 기본 URL
crew_bearer_token	`str`	Yes	None	API 인증용 bearer token
crew_name	`str`	Yes	None	크루 자동화의 이름
crew_description	`str`	Yes	None	크루 자동화가 하는 일에 대한 설명
max_polling_time	`int`	No	600	태스크 완료를 기다리는 최대 시간(초)
crew_inputs	`dict`	No	None	커스텀 입력 스키마 필드를 정의하는 딕셔너리

환경변수 (Environment Variables)

CREWAI_API_URL=https://your-crew-automation.crewai.com  # crew_api_url 전달의 대안
CREWAI_BEARER_TOKEN=your_bearer_token_here              # crew_bearer_token 전달의 대안

고급 사용법 (Advanced Usage)

동적 파라미터가 있는 커스텀 입력 스키마

from crewai import Agent, Task, Crew
from crewai_tools import InvokeCrewAIAutomationTool
from pydantic import Field

# 커스텀 입력 스키마 정의
custom_inputs = {
    "year": Field(..., description="Year to retrieve the report for (integer)"),
    "region": Field(default="global", description="Geographic region for analysis"),
    "format": Field(default="summary", description="Report format (summary, detailed, raw)")
}

# 커스텀 입력으로 도구 생성
market_research_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://state-of-ai-report-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="State of AI Report",
    crew_description="Retrieves a comprehensive report on state of AI for a given year and region",
    crew_inputs=custom_inputs,
    max_polling_time=15 * 60  # 15분 타임아웃
)

# 이 도구를 사용하는 에이전트 생성
research_agent = Agent(
    role="Research Coordinator",
    goal="Coordinate and execute market research tasks",
    backstory="You are an expert at coordinating research tasks and leveraging automation tools.",
    tools=[market_research_tool],
    verbose=True
)

# 커스텀 파라미터로 태스크 생성 및 실행
research_task = Task(
    description="Conduct market research on AI tools market for 2024 in North America with detailed format",
    agent=research_agent,
    expected_output="Comprehensive market research report"
)

crew = Crew(
    agents=[research_agent],
    tasks=[research_task]
)

result = crew.kickoff()

멀티 스테이지 자동화 워크플로

from crewai import Agent, Task, Crew, Process
from crewai_tools import InvokeCrewAIAutomationTool

# 서로 다른 자동화 도구 초기화
data_collection_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://data-collection-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Data Collection Automation",
    crew_description="Collects and preprocesses raw data"
)

analysis_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://analysis-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Analysis Automation",
    crew_description="Performs advanced data analysis and modeling"
)

reporting_tool = InvokeCrewAIAutomationTool(
    crew_api_url="https://reporting-crew-[...].crewai.com",
    crew_bearer_token="your_bearer_token_here",
    crew_name="Reporting Automation",
    crew_description="Generates comprehensive reports and visualizations"
)

# 전문화된 에이전트 생성
data_collector = Agent(
    role='Data Collection Specialist',
    goal='Gather and preprocess data from various sources',
    backstory='I specialize in collecting and cleaning data from multiple sources.',
    tools=[data_collection_tool]
)

data_analyst = Agent(
    role='Data Analysis Expert',
    goal='Perform advanced analysis on collected data',
    backstory='I am an expert in statistical analysis and machine learning.',
    tools=[analysis_tool]
)

report_generator = Agent(
    role='Report Generation Specialist',
    goal='Create comprehensive reports and visualizations',
    backstory='I excel at creating clear, actionable reports from complex data.',
    tools=[reporting_tool]
)

# 순차 태스크 생성
collection_task = Task(
    description="Collect market data for Q4 2024 analysis",
    agent=data_collector
)

analysis_task = Task(
    description="Analyze collected data to identify trends and patterns",
    agent=data_analyst
)

reporting_task = Task(
    description="Generate executive summary report with key insights and recommendations",
    agent=report_generator
)

# 순차 처리로 크루 생성
crew = Crew(
    agents=[data_collector, data_analyst, report_generator],
    tasks=[collection_task, analysis_task, reporting_task],
    process=Process.sequential,
    verbose=2
)

result = crew.kickoff()

사용 사례 (Use Cases)

분산 크루 오케스트레이션 (Distributed Crew Orchestration)

복잡한 멀티 스테이지 워크플로를 처리하기 위해 여러 전문화된 크루 자동화 조율하기
종합적인 태스크 실행을 위해 서로 다른 자동화 서비스 간의 원활한 핸드오프 가능하게 하기
여러 CrewAI Platform 자동화에 워크로드를 분산해 처리 확장하기

크로스 플랫폼 통합 (Cross-Platform Integration)

하이브리드 로컬-클라우드 워크플로를 위해 CrewAI 에이전트와 CrewAI Platform 자동화 연결하기
로컬 제어와 오케스트레이션을 유지하면서 전문화된 자동화 활용하기
로컬 에이전트와 클라우드 기반 자동화 서비스 간의 안전한 협업 구현하기

엔터프라이즈 자동화 파이프라인 (Enterprise Automation Pipelines)

로컬 지능과 클라우드 처리 능력을 결합한 엔터프라이즈급 자동화 파이프라인 구축하기
여러 자동화 서비스에 걸친 복잡한 비즈니스 워크플로 구현하기
데이터 분석, 보고, 의사결정을 위한 확장 가능하고 반복 가능한 프로세스 만들기

동적 워크플로 구성 (Dynamic Workflow Composition)

태스크 요구사항에 따라 서로 다른 자동화 서비스를 연결해 워크플로를 동적으로 구성하기
데이터 특성이나 비즈니스 규칙에 따라 자동화 선택이 달라지는 적응형 처리 구현하기
다양한 방식으로 결합할 수 있는 유연하고 재사용 가능한 자동화 컴포넌트 만들기

전문 도메인 처리 (Specialized Domain Processing)

범용 에이전트에서 도메인 특화 자동화(금융 분석, 법률 리서치, 기술 문서)에 접근하기
복잡한 도메인 로직을 다시 만들지 않고 사전 구축된 전문화된 크루 자동화 활용하기
타겟 자동화 서비스를 통해 에이전트가 전문가 수준의 기능에 접근할 수 있게 하기

커스텀 입력 스키마 (Custom Input Schema)

crew_inputs를 정의할 때는 Pydantic Field 객체를 사용해 입력 파라미터를 지정합니다.

from pydantic import Field

crew_inputs = {
    "required_param": Field(..., description="This parameter is required"),
    "optional_param": Field(default="default_value", description="This parameter is optional"),
    "typed_param": Field(..., description="Integer parameter", ge=1, le=100)  # 검증 포함
}

에러 처리 (Error Handling)

이 도구는 일반적인 시나리오에 대한 종합적인 에러 처리를 제공합니다.

API 연결 오류: CrewAI Platform과의 네트워크 연결 문제
인증 오류: 잘못되었거나 만료된 bearer token
타임아웃 오류: 최대 폴링 시간을 초과하는 태스크
태스크 실패: 실행 중 실패하는 크루 자동화
입력 검증 오류: 자동화 엔드포인트에 전달된 잘못된 파라미터

API 엔드포인트 (API Endpoints)

이 도구는 두 개의 주요 API 엔드포인트와 상호작용합니다.

POST {crew_api_url}/kickoff: 새 크루 자동화 태스크를 시작합니다
GET {crew_api_url}/status/{crew_id}: 실행 중인 태스크의 상태를 확인합니다

참고 사항 (Notes)

이 도구는 완료 또는 타임아웃까지 매초 상태 엔드포인트를 자동으로 폴링합니다
성공한 태스크는 결과를 직접 반환하고, 실패한 태스크는 에러 정보를 반환합니다
bearer token은 안전하게 보관해야 하며 프로덕션 환경에서 하드코딩하지 마세요
bearer token 같은 민감한 설정은 환경변수를 사용하는 것을 고려하세요
커스텀 입력 스키마는 대상 크루 자동화가 기대하는 파라미터와 호환되어야 합니다

CrewAI Run Automation 도구