​
개요

CrewAI는 강력한 이벤트 시스템을 제공하여 crew 실행 중 발생하는 다양한 이벤트를 수신하고 이에 반응할 수 있도록 합니다. 이 기능을 통해 맞춤형 통합, 모니터링 솔루션, 로깅 시스템 또는 CrewAI의 내부 이벤트에 따라 트리거되어야 하는 기타 모든 기능을 구축할 수 있습니다.

​
작동 방식

CrewAI는 실행 수명 주기 전반에 걸쳐 이벤트를 발생시키는 이벤트 버스 아키텍처를 사용합니다. 이벤트 시스템은 다음과 같은 구성 요소로 구축되어 있습니다:
  1. CrewAIEventsBus: 이벤트 등록 및 발생을 관리하는 싱글톤 이벤트 버스
  2. BaseEvent: 시스템 내 모든 이벤트의 기본 클래스
  3. BaseEventListener: 커스텀 이벤트 리스너 생성을 위한 추상 기본 클래스
CrewAI에서 특정 동작(예: Crew가 실행을 시작하거나 Agent가 task를 완료하거나 tool이 사용될 때)이 발생하면, 시스템은 해당 이벤트를 발생시킵니다. 이러한 이벤트에 대한 핸들러를 등록하여 해당 이벤트가 발생할 때 커스텀 코드를 실행할 수 있습니다.
CrewAI Enterprise는 event 시스템을 활용하여 모든 prompt, completion 및 관련 메타데이터를 추적, 저장 및 시각화하는 내장 Prompt Tracing 기능을 제공합니다. 이 기능을 통해 agent 운영에 대한 강력한 디버깅 기능과 투명성을 얻을 수 있습니다.Prompt Tracing DashboardPrompt Tracing을 통해 다음과 같은 작업이 가능합니다:
  • LLM에 전송된 모든 prompt의 전체 기록 보기
  • token 사용량 및 비용 추적
  • agent reasoning 실패 디버깅
  • 팀 내에서 prompt 시퀀스 공유
  • 다양한 prompt 전략 비교
  • 컴플라이언스 및 감사를 위한 trace 내보내기

​
커스텀 이벤트 리스너 생성하기

커스텀 이벤트 리스너를 생성하려면 다음 단계를 따라야 합니다:
  1. BaseEventListener를 상속하는 클래스를 생성합니다.
  2. setup_listeners 메서드를 구현합니다.
  3. 원하는 이벤트에 대한 핸들러를 등록합니다.
  4. 해당 파일에서 리스너의 인스턴스를 생성합니다.
아래는 커스텀 이벤트 리스너 클래스의 간단한 예시입니다: 
from crewai.utilities.events import (
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener

class MyCustomListener(BaseEventListener):
    def __init__(self):
        super().__init__()

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_started(source, event):
            print(f"Crew '{event.crew_name}' has started execution!")

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_crew_completed(source, event):
            print(f"Crew '{event.crew_name}' has completed execution!")
            print(f"Output: {event.output}")

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_execution_completed(source, event):
            print(f"Agent '{event.agent.role}' completed task")
            print(f"Output: {event.output}")

​
리스너를 올바르게 등록하기

리스너 클래스를 정의하는 것만으로는 충분하지 않습니다. 해당 클래스의 인스턴스를 생성하고 애플리케이션에 임포트되었는지 확인해야 합니다. 이렇게 하면 다음과 같은 효과가 있습니다:
  1. 이벤트 핸들러가 이벤트 버스에 등록됩니다.
  2. 리스너 인스턴스가 메모리에 유지됩니다(가비지 컬렉션되지 않음).
  3. 이벤트가 발생할 때 리스너가 활성화됩니다.

​
옵션 1: Crew 또는 Flow 구현에서 가져오기 및 인스턴스화

가장 중요한 것은 Crew 또는 Flow가 정의되고 실행되는 파일에서 리스너의 인스턴스를 생성하는 것입니다.

​
크루 기반 애플리케이션의 경우

크루 구현 파일 상단에 리스너를 생성하고 임포트하세요: 
# In your crew.py file
from crewai import Agent, Crew, Task
from my_listeners import MyCustomListener

# Create an instance of your listener
my_listener = MyCustomListener()

class MyCustomCrew:
    # Your crew implementation...

    def crew(self):
        return Crew(
            agents=[...],
            tasks=[...],
            # ...
        )

​
플로우 기반 애플리케이션의 경우

플로우 구현 파일 상단에 리스너를 생성하고 임포트하세요: 
# main.py 또는 flow.py 파일에서
from crewai.flow import Flow, listen, start
from my_listeners import MyCustomListener

# 리스너 인스턴스 생성
my_listener = MyCustomListener()

class MyCustomFlow(Flow):
    # 플로우 구현...

    @start()
    def first_step(self):
        # ...
이렇게 하면 크루 또는 플로우가 실행될 때 리스너가 로드되고 활성화됩니다.

​
옵션 2: 리스너를 위한 패키지 생성

여러 개의 리스너가 있는 경우 등 보다 구조적인 접근 방식을 원한다면 다음과 같이 진행하세요:
  1. 리스너를 위한 패키지를 생성합니다:
my_project/
  ├── listeners/
  │   ├── __init__.py
  │   ├── my_custom_listener.py
  │   └── another_listener.py
  1. my_custom_listener.py에서 리스너 클래스를 정의하고 인스턴스를 생성합니다:
# my_custom_listener.py
from crewai.utilities.events.base_event_listener import BaseEventListener
# ... import events ...

class MyCustomListener(BaseEventListener):
    # ... implementation ...

# 리스너 인스턴스 생성
my_custom_listener = MyCustomListener()
  1. __init__.py에서 리스너 인스턴스를 임포트하여 로드되도록 합니다:
# __init__.py
from .my_custom_listener import my_custom_listener
from .another_listener import another_listener

# 다른 곳에서 접근이 필요하다면 익스포트할 수도 있습니다
__all__ = ['my_custom_listener', 'another_listener']
  1. Crew나 Flow 파일에서 리스너 패키지를 임포트합니다:
# crew.py 또는 flow.py 파일 내에서
import my_project.listeners  # 모든 리스너가 로드됩니다

class MyCustomCrew:
    # Your crew implementation...
이것이 바로 CrewAI의 내장 agentops_listener가 등록되는 방식과 동일합니다. CrewAI 코드베이스에서는 다음과 같이 되어 있습니다: 
# src/crewai/utilities/events/third_party/__init__.py
from .agentops_listener import agentops_listener
이렇게 하면 crewai.utilities.events 패키지가 임포트될 때 agentops_listener가 자동으로 로드됩니다.

​
사용 가능한 이벤트 유형

CrewAI는 여러분이 청취할 수 있는 다양한 이벤트를 제공합니다:

​
Crew 이벤트

  • CrewKickoffStartedEvent: Crew가 실행을 시작할 때 발생
  • CrewKickoffCompletedEvent: Crew가 실행을 완료할 때 발생
  • CrewKickoffFailedEvent: Crew가 실행을 완료하지 못할 때 발생
  • CrewTestStartedEvent: Crew가 테스트를 시작할 때 발생
  • CrewTestCompletedEvent: Crew가 테스트를 완료할 때 발생
  • CrewTestFailedEvent: Crew가 테스트를 완료하지 못할 때 발생
  • CrewTrainStartedEvent: Crew가 훈련을 시작할 때 발생
  • CrewTrainCompletedEvent: Crew가 훈련을 완료할 때 발생
  • CrewTrainFailedEvent: Crew가 훈련을 완료하지 못할 때 발생

​
에이전트 이벤트

  • AgentExecutionStartedEvent: 에이전트가 작업 실행을 시작할 때 발생함
  • AgentExecutionCompletedEvent: 에이전트가 작업 실행을 완료할 때 발생함
  • AgentExecutionErrorEvent: 에이전트가 실행 도중 오류를 만날 때 발생함

​
작업 이벤트

  • TaskStartedEvent: 작업이 실행을 시작할 때 발생
  • TaskCompletedEvent: 작업이 실행을 완료할 때 발생
  • TaskFailedEvent: 작업이 실행을 완료하지 못할 때 발생
  • TaskEvaluationEvent: 작업이 평가될 때 발생

​
도구 사용 이벤트

  • ToolUsageStartedEvent: 도구 실행이 시작될 때 발생함
  • ToolUsageFinishedEvent: 도구 실행이 완료될 때 발생함
  • ToolUsageErrorEvent: 도구 실행 중 오류가 발생할 때 발생함
  • ToolValidateInputErrorEvent: 도구 입력 검증 중 오류가 발생할 때 발생함
  • ToolExecutionErrorEvent: 도구 실행 중 오류가 발생할 때 발생함
  • ToolSelectionErrorEvent: 도구 선택 시 오류가 발생할 때 발생함

​
지식 이벤트

  • KnowledgeRetrievalStartedEvent: 지식 검색이 시작될 때 발생
  • KnowledgeRetrievalCompletedEvent: 지식 검색이 완료될 때 발생
  • KnowledgeQueryStartedEvent: 지식 쿼리가 시작될 때 발생
  • KnowledgeQueryCompletedEvent: 지식 쿼리가 완료될 때 발생
  • KnowledgeQueryFailedEvent: 지식 쿼리가 실패할 때 발생
  • KnowledgeSearchQueryFailedEvent: 지식 검색 쿼리가 실패할 때 발생

​
LLM 가드레일 이벤트

  • LLMGuardrailStartedEvent: 가드레일 검증이 시작될 때 발생합니다. 적용되는 가드레일에 대한 세부 정보와 재시도 횟수를 포함합니다.
  • LLMGuardrailCompletedEvent: 가드레일 검증이 완료될 때 발생합니다. 검증의 성공/실패, 결과 및 오류 메시지(있는 경우)에 대한 세부 정보를 포함합니다.

​
Flow 이벤트

  • FlowCreatedEvent: Flow가 생성될 때 발생
  • FlowStartedEvent: Flow가 실행을 시작할 때 발생
  • FlowFinishedEvent: Flow가 실행을 완료할 때 발생
  • FlowPlotEvent: Flow가 플롯될 때 발생
  • MethodExecutionStartedEvent: Flow 메서드가 실행을 시작할 때 발생
  • MethodExecutionFinishedEvent: Flow 메서드가 실행을 완료할 때 발생
  • MethodExecutionFailedEvent: Flow 메서드가 실행을 완료하지 못할 때 발생

​
LLM 이벤트

  • LLMCallStartedEvent: LLM 호출이 시작될 때 발생
  • LLMCallCompletedEvent: LLM 호출이 완료될 때 발생
  • LLMCallFailedEvent: LLM 호출이 실패할 때 발생
  • LLMStreamChunkEvent: 스트리밍 LLM 응답 중 각 청크를 받을 때마다 발생

​
메모리 이벤트

  • MemoryQueryStartedEvent: 메모리 쿼리가 시작될 때 발생합니다. 쿼리, limit, 선택적 score threshold를 포함합니다.
  • MemoryQueryCompletedEvent: 메모리 쿼리가 성공적으로 완료될 때 발생합니다. 쿼리, 결과, limit, score threshold, 쿼리 실행 시간을 포함합니다.
  • MemoryQueryFailedEvent: 메모리 쿼리 실행에 실패할 때 발생합니다. 쿼리, limit, score threshold, 오류 메시지를 포함합니다.
  • MemorySaveStartedEvent: 메모리 저장 작업이 시작될 때 발생합니다. 저장할 값, 메타데이터, 선택적 agent 역할을 포함합니다.
  • MemorySaveCompletedEvent: 메모리 저장 작업이 성공적으로 완료될 때 발생합니다. 저장된 값, 메타데이터, agent 역할, 저장 실행 시간을 포함합니다.
  • MemorySaveFailedEvent: 메모리 저장 작업에 실패할 때 발생합니다. 값, 메타데이터, agent 역할, 오류 메시지를 포함합니다.
  • MemoryRetrievalStartedEvent: 태스크 프롬프트를 위한 메모리 검색이 시작될 때 발생합니다. 선택적 태스크 ID를 포함합니다.
  • MemoryRetrievalCompletedEvent: 태스크 프롬프트를 위한 메모리 검색이 성공적으로 완료될 때 발생합니다. 태스크 ID, 메모리 내용, 검색 실행 시간을 포함합니다.

​
이벤트 핸들러 구조

각 이벤트 핸들러는 두 개의 매개변수를 받습니다:
  1. source: 이벤트를 발생시킨 객체
  2. event: 이벤트별 데이터를 포함하는 이벤트 인스턴스
이벤트 객체의 구조는 이벤트 타입에 따라 다르지만, 모든 이벤트는 BaseEvent를 상속하며 다음을 포함합니다:
  • timestamp: 이벤트가 발생한 시간
  • type: 이벤트 타입을 나타내는 문자열 식별자
추가 필드는 이벤트 타입에 따라 다릅니다. 예를 들어, CrewKickoffCompletedEvent에는 crew_name과 output 필드가 포함됩니다.

​
실제 예시: AgentOps와의 통합

CrewAI는 AI 에이전트를 위한 모니터링 및 관찰 플랫폼인 AgentOps와의 서드파티 통합 예시를 포함하고 있습니다. 구현 방식은 다음과 같습니다: 
from typing import Optional

from crewai.utilities.events import (
    CrewKickoffCompletedEvent,
    ToolUsageErrorEvent,
    ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent

try:
    import agentops
    AGENTOPS_INSTALLED = True
except ImportError:
    AGENTOPS_INSTALLED = False

class AgentOpsListener(BaseEventListener):
    tool_event: Optional["agentops.ToolEvent"] = None
    session: Optional["agentops.Session"] = None

    def __init__(self):
        super().__init__()

    def setup_listeners(self, crewai_event_bus):
        if not AGENTOPS_INSTALLED:
            return

        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
            self.session = agentops.init()
            for agent in source.agents:
                if self.session:
                    self.session.create_agent(
                        name=agent.role,
                        agent_id=str(agent.id),
                    )

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
            if self.session:
                self.session.end_session(
                    end_state="Success",
                    end_state_reason="Finished Execution",
                )

        @crewai_event_bus.on(ToolUsageStartedEvent)
        def on_tool_usage_started(source, event: ToolUsageStartedEvent):
            self.tool_event = agentops.ToolEvent(name=event.tool_name)
            if self.session:
                self.session.record(self.tool_event)

        @crewai_event_bus.on(ToolUsageErrorEvent)
        def on_tool_usage_error(source, event: ToolUsageErrorEvent):
            agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)
이 listener는 crew가 시작될 때 AgentOps 세션을 초기화하고, agent를 AgentOps에 등록하며, 도구 사용을 추적하고, crew가 완료되면 세션을 종료합니다. AgentOps listener는 src/crewai/utilities/events/third_party/__init__.py 파일의 import를 통해 CrewAI 이벤트 시스템에 등록됩니다: 
from .agentops_listener import agentops_listener
이렇게 하면 crewai.utilities.events 패키지가 import될 때 agentops_listener가 로드되는 것이 보장됩니다.

​
고급 사용법: Scoped Handlers

임시 이벤트 처리가 필요한 경우(테스트 또는 특정 작업에 유용함), scoped_handlers 컨텍스트 관리자를 사용할 수 있습니다: 
from crewai.utilities.events import crewai_event_bus, CrewKickoffStartedEvent

with crewai_event_bus.scoped_handlers():
    @crewai_event_bus.on(CrewKickoffStartedEvent)
    def temp_handler(source, event):
        print("This handler only exists within this context")

    # Do something that emits events

# 컨텍스트 밖에서는 임시 핸들러가 제거됩니다

​
사용 사례

이벤트 리스너는 다양한 목적으로 사용할 수 있습니다:
  1. 로깅 및 모니터링: Crew의 실행을 추적하고 중요한 이벤트를 기록합니다
  2. 분석: Crew의 성능과 동작에 대한 데이터를 수집합니다
  3. 디버깅: 특정 문제를 디버깅하기 위해 임시 리스너를 설정합니다
  4. 통합: CrewAI를 모니터링 플랫폼, 데이터베이스 또는 알림 서비스와 같은 외부 시스템과 연결합니다
  5. 사용자 정의 동작: 특정 이벤트에 따라 사용자 정의 동작을 트리거합니다

​
모범 사례

  1. 핸들러를 가볍게 유지하세요: 이벤트 핸들러는 경량이어야 하며, 블로킹 작업을 피해야 합니다.
  2. 오류 처리: 예외가 메인 실행에 영향을 주지 않도록 이벤트 핸들러에 적절한 오류 처리를 포함하세요.
  3. 정리: 리스너가 자원을 할당한다면, 이를 적절하게 정리하는지 확인하세요.
  4. 선택적 리스닝: 실제로 처리해야 하는 이벤트에만 리스닝하세요.
  5. 테스트: 이벤트 리스너가 예상대로 동작하는지 독립적으로 테스트하세요.
CrewAI의 이벤트 시스템을 활용하면 기능을 확장하고 기존 인프라와 원활하게 통합할 수 있습니다.