이벤트 리스너

​

개요

​

작동 방식

1. CrewAIEventsBus: 이벤트 등록 및 발생을 관리하는 싱글톤 이벤트 버스
2. BaseEvent: 시스템 내 모든 이벤트의 기본 클래스
3. BaseEventListener: 커스텀 이벤트 리스너 생성을 위한 추상 기본 클래스

CrewAI AMP는 event 시스템을 활용하여 모든 prompt, completion 및 관련 메타데이터를 추적, 저장 및 시각화하는 내장 Prompt Tracing 기능을 제공합니다. 이 기능을 통해 agent 운영에 대한 강력한 디버깅 기능과 투명성을 얻을 수 있습니다.Prompt Tracing을 통해 다음과 같은 작업이 가능합니다:

• LLM에 전송된 모든 prompt의 전체 기록 보기
• token 사용량 및 비용 추적
• agent reasoning 실패 디버깅
• 팀 내에서 prompt 시퀀스 공유
• 다양한 prompt 전략 비교
• 컴플라이언스 및 감사를 위한 trace 내보내기

​

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

1. BaseEventListener를 상속하는 클래스를 생성합니다.
2. setup_listeners 메서드를 구현합니다.
3. 원하는 이벤트에 대한 핸들러를 등록합니다.
4. 해당 파일에서 리스너의 인스턴스를 생성합니다.

from crewai.events import (
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)
from crewai.events 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 구현에서 가져오기 및 인스턴스화

​

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

# 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

2. my_custom_listener.py에서 리스너 클래스를 정의하고 인스턴스를 생성합니다:

# my_custom_listener.py
from crewai.events import BaseEventListener
# ... import events ...

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

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

3. __init__.py에서 리스너 인스턴스를 임포트하여 로드되도록 합니다:

# __init__.py
from .my_custom_listener import my_custom_listener
from .another_listener import another_listener

# 다른 곳에서 접근이 필요하다면 익스포트할 수도 있습니다
__all__ = ['my_custom_listener', 'another_listener']

4. Crew나 Flow 파일에서 리스너 패키지를 임포트합니다:

# crew.py 또는 flow.py 파일 내에서
import my_project.listeners  # 모든 리스너가 로드됩니다

class MyCustomCrew:
    # Your crew implementation...

​

사용 가능한 이벤트 유형

​

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: 이벤트별 데이터를 포함하는 이벤트 인스턴스

• timestamp: 이벤트가 발생한 시간
• type: 이벤트 타입을 나타내는 문자열 식별자

​

고급 사용법: Scoped Handlers

from crewai.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. 테스트: 이벤트 리스너가 예상대로 동작하는지 독립적으로 테스트하세요.