Listeners de Evento

​

Visão Geral

​

Como Funciona

1. CrewAIEventsBus: Um event bus singleton que gerencia o registro e emissão de eventos
2. BaseEvent: Classe base para todos os eventos do sistema
3. BaseEventListener: Classe base abstrata para criar listeners de evento personalizados

O CrewAI Enterprise fornece o recurso Prompt Tracing, que aproveita o sistema de eventos para rastrear, armazenar e visualizar todos os prompts, respostas e metadados associados. Isso proporciona poderosas capacidades de depuração e transparência nas operações dos seus agentes.Com o Prompt Tracing você pode:

• Visualizar o histórico completo de todos os prompts enviados ao seu LLM
• Monitorar o uso de tokens e custos
• Depurar falhas de raciocínio dos agentes
• Compartilhar sequências de prompts com sua equipe
• Comparar diferentes estratégias de prompts
• Exportar rastreamentos para compliance e auditoria

​

Criando um Listener de Evento Personalizado

1. Criar uma classe que herde de BaseEventListener
2. Implementar o método setup_listeners
3. Registrar handles para os eventos de seu interesse
4. Instanciar seu listener no arquivo apropriado

from crewai.utilities.events import (
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener

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

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def ao_iniciar_crew(source, event):
            print(f"Crew '{event.crew_name}' iniciou a execução!")

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def ao_finalizar_crew(source, event):
            print(f"Crew '{event.crew_name}' finalizou a execução!")
            print(f"Saída: {event.output}")

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def ao_finalizar_execucao_agente(source, event):
            print(f"Agente '{event.agent.role}' concluiu a tarefa")
            print(f"Saída: {event.output}")

​

Registrando Corretamente Seu Listener

1. Os event handlers estejam registrados no event bus
2. A instância do listener permaneça em memória (não seja coletada pelo garbage collector)
3. O listener esteja ativo quando os eventos forem emitidos

​

Opção 1: Importar e Instanciar no Seu Crew ou Implementação de Flow

​

Para Aplicações Baseadas em Crew

# No seu arquivo crew.py
from crewai import Agent, Crew, Task
from my_listeners import MyCustomListener

# Crie uma instância do seu listener
my_listener = MyCustomListener()

class MyCustomCrew:
    # Sua implementação do crew...

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

​

Para Aplicações Baseadas em Flow

# Em seu arquivo main.py ou flow.py
from crewai.flow import Flow, listen, start
from my_listeners import MyCustomListener

# Crie uma instância do seu listener
my_listener = MyCustomListener()

class MyCustomFlow(Flow):
    # Sua implementação do flow...

    @start()
    def first_step(self):
        # ...

​

Opção 2: Criar um Pacote para Seus Listeners

1. Crie um pacote para seus listeners:

my_project/
  ├── listeners/
  │   ├── __init__.py
  │   ├── my_custom_listener.py
  │   └── another_listener.py

2. Em my_custom_listener.py, defina sua classe de listener e crie uma instância:

# my_custom_listener.py
from crewai.utilities.events.base_event_listener import BaseEventListener
# ... importe events ...

class MyCustomListener(BaseEventListener):
    # ... implementação ...

# Crie uma instância do seu listener
my_custom_listener = MyCustomListener()

3. Em __init__.py, importe as instâncias dos listeners para garantir seu carregamento:

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

# Opcionalmente exporte-os se precisar acessá-los em outros lugares
__all__ = ['my_custom_listener', 'another_listener']

4. Importe seu pacote de listeners no arquivo do seu Crew ou Flow:

# No seu arquivo crew.py ou flow.py
import my_project.listeners  # Isso carrega todos os seus listeners

class MyCustomCrew:
    # Sua implementação do crew...

​

Tipos de Eventos Disponíveis

​

Eventos de Crew

• CrewKickoffStartedEvent: Emitido quando um Crew inicia a execução
• CrewKickoffCompletedEvent: Emitido quando um Crew conclui a execução
• CrewKickoffFailedEvent: Emitido quando um Crew falha ao concluir a execução
• CrewTestStartedEvent: Emitido ao iniciar o teste de um Crew
• CrewTestCompletedEvent: Emitido ao concluir o teste de um Crew
• CrewTestFailedEvent: Emitido ao falhar no teste de um Crew
• CrewTrainStartedEvent: Emitido ao iniciar o treinamento de um Crew
• CrewTrainCompletedEvent: Emitido ao concluir o treinamento de um Crew
• CrewTrainFailedEvent: Emitido ao falhar no treinamento de um Crew

​

Eventos de Agent

• AgentExecutionStartedEvent: Emitido quando um Agent inicia a execução de uma tarefa
• AgentExecutionCompletedEvent: Emitido quando um Agent conclui a execução de uma tarefa
• AgentExecutionErrorEvent: Emitido quando um Agent encontra um erro durante a execução

​

Eventos de Task

• TaskStartedEvent: Emitido ao iniciar a execução de uma Task
• TaskCompletedEvent: Emitido ao concluir a execução de uma Task
• TaskFailedEvent: Emitido ao falhar na execução de uma Task
• TaskEvaluationEvent: Emitido quando uma Task é avaliada

​

Eventos de Uso de Ferramentas

• ToolUsageStartedEvent: Emitido ao iniciar a execução de uma ferramenta
• ToolUsageFinishedEvent: Emitido ao concluir a execução de uma ferramenta
• ToolUsageErrorEvent: Emitido quando ocorre erro na execução de uma ferramenta
• ToolValidateInputErrorEvent: Emitido ao ocorrer erro de validação de entrada na ferramenta
• ToolExecutionErrorEvent: Emitido quando ocorre erro na execução de uma ferramenta
• ToolSelectionErrorEvent: Emitido ao ocorrer erro na seleção de uma ferramenta

​

Eventos de Knowledge

• KnowledgeRetrievalStartedEvent: Emitido ao iniciar recuperação de conhecimento
• KnowledgeRetrievalCompletedEvent: Emitido ao concluir recuperação de conhecimento
• KnowledgeQueryStartedEvent: Emitido ao iniciar consulta de conhecimento
• KnowledgeQueryCompletedEvent: Emitido ao concluir consulta de conhecimento
• KnowledgeQueryFailedEvent: Emitido ao falhar consulta de conhecimento
• KnowledgeSearchQueryFailedEvent: Emitido ao falhar consulta de busca de conhecimento

​

Eventos de Guardrail do LLM

• LLMGuardrailStartedEvent: Emitido ao iniciar validação dos guardrails. Contém detalhes do guardrail aplicado e tentativas.
• LLMGuardrailCompletedEvent: Emitido ao concluir validação dos guardrails. Contém detalhes sobre sucesso/falha na validação, resultados e mensagens de erro, se houver.

​

Eventos de Flow

• FlowCreatedEvent: Emitido ao criar um Flow
• FlowStartedEvent: Emitido ao iniciar a execução de um Flow
• FlowFinishedEvent: Emitido ao concluir a execução de um Flow
• FlowPlotEvent: Emitido ao plotar um Flow
• MethodExecutionStartedEvent: Emitido ao iniciar a execução de um método do Flow
• MethodExecutionFinishedEvent: Emitido ao concluir a execução de um método do Flow
• MethodExecutionFailedEvent: Emitido ao falhar na execução de um método do Flow

​

Eventos de LLM

• LLMCallStartedEvent: Emitido ao iniciar uma chamada LLM
• LLMCallCompletedEvent: Emitido ao concluir uma chamada LLM
• LLMCallFailedEvent: Emitido ao falhar uma chamada LLM
• LLMStreamChunkEvent: Emitido para cada chunk recebido durante respostas em streaming do LLM

​

Estrutura dos Handlers de Evento

1. source: O objeto que emitiu o evento
2. event: A instância do evento, contendo dados específicos do evento

• timestamp: O horário em que o evento foi emitido
• type: Identificador do tipo do evento

​

Uso Avançado: Handlers Escopados

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("Este handler só existe neste contexto")

    # Faça algo que emita eventos

# Fora do contexto, o handler temporário é removido

​

Casos de Uso

1. Log e Monitoramento: Monitore a execução do seu Crew e registre eventos importantes
2. Analytics: Colete dados sobre o desempenho e comportamento do seu Crew
3. Depuração: Configure listeners temporários para debugar problemas específicos
4. Integração: Conecte o CrewAI a sistemas externos como plataformas de monitoramento, bancos de dados ou serviços de notificação
5. Comportamento Personalizado: Dispare ações personalizadas com base em eventos específicos

​

Boas Práticas

1. Mantenha Handlers Leves: Handlers de eventos devem ser leves e evitar operações bloqueantes
2. Tratamento de Erros: Implemente tratamento de erros adequado nos event handlers para evitar que exceções afetem a execução principal
3. Limpeza: Se seu listener alocar recursos, garanta o devido fechamento/liberação
4. Escuta Seletiva: Escute apenas eventos que realmente precisa tratar
5. Testes: Teste seus listeners de evento isoladamente para garantir que se comportam conforme esperado