事件监听器

概览

CrewAI 提供了一个强大的事件系统，让你可以监听并响应 Crew 执行期间发生的各种事件。此功能可让你构建自定义集成、监控方案、日志系统，或任何需要基于 CrewAI 内部事件触发的功能。

工作原理

CrewAI 使用事件总线架构在整个执行生命周期中发出事件。该事件系统由以下组件构成：

CrewAIEventsBus：负责管理事件注册与发出的单例事件总线
BaseEvent：系统中所有事件的基类
BaseEventListener：用于创建自定义事件监听器的抽象基类

当 CrewAI 中发生特定动作时（例如 Crew 开始执行、Agent 完成任务，或某个工具被使用），系统会发出相应事件。你可以为这些事件注册处理器，以便在事件发生时执行自定义代码。

创建自定义事件监听器

要创建自定义事件监听器，你需要：

创建一个继承自 BaseEventListener 的类
实现 setup_listeners 方法
为你感兴趣的事件注册处理器
在合适的文件中创建该监听器的实例

下面是一个简单的自定义事件监听器类示例：

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：在 Crew 或 Flow 实现中导入并实例化

最重要的是，在定义和执行 Crew 或 Flow 的文件中创建监听器实例：

对于基于 Crew 的应用

在 Crew 实现文件顶部创建并导入监听器：

# 在你的 crew.py 文件中
from crewai import Agent, Crew, Task
from my_listeners import MyCustomListener

# 创建监听器实例
my_listener = MyCustomListener()

class MyCustomCrew:
    # 你的 crew 实现...

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

对于基于 Flow 的应用

在 Flow 实现文件顶部创建并导入监听器：

# 在你的 main.py 或 flow.py 文件中
from crewai.flow import Flow, listen, start
from my_listeners import MyCustomListener

# 创建监听器实例
my_listener = MyCustomListener()

class MyCustomFlow(Flow):
    # 你的 flow 实现...

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

这样可以确保在 Crew 或 Flow 执行时，监听器已加载并处于激活状态。

方案 2：为监听器创建一个包

如果你有多个监听器，采用更结构化的方式会更好：

为监听器创建一个包：

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

在 my_custom_listener.py 中定义监听器类并创建其实例：

from crewai.events import BaseEventListener
# ... 导入事件 ...

class MyCustomListener(BaseEventListener):
    # ... 实现 ...

# 创建监听器实例
my_custom_listener = MyCustomListener()

在 __init__.py 中导入监听器实例，确保它们会被加载：

from .my_custom_listener import my_custom_listener
from .another_listener import another_listener

# 如果你需要在其他地方访问，也可以选择导出它们
__all__ = ['my_custom_listener', 'another_listener']

在 Crew 或 Flow 文件中导入你的监听器包：

# 在你的 crew.py 或 flow.py 文件中
import my_project.listeners  # 这会加载所有监听器

class MyCustomCrew:
    # 你的 crew 实现...

这就是 CrewAI 代码库中第三方事件监听器的注册方式。

可用的事件类型

CrewAI 提供了大量可供监听的事件：

Crew 事件

CrewKickoffStartedEvent：当 Crew 开始执行时发出
CrewKickoffCompletedEvent：当 Crew 完成执行时发出
CrewKickoffFailedEvent：当 Crew 执行失败时发出
CrewTestStartedEvent：当 Crew 开始测试时发出
CrewTestCompletedEvent：当 Crew 完成测试时发出
CrewTestFailedEvent：当 Crew 测试失败时发出
CrewTrainStartedEvent：当 Crew 开始训练时发出
CrewTrainCompletedEvent：当 Crew 完成训练时发出
CrewTrainFailedEvent：当 Crew 训练失败时发出
CrewTestResultEvent：当 Crew 测试结果可用时发出。包含质量分数、执行时长和所用模型。

Agent 事件

AgentExecutionStartedEvent：当 Agent 开始执行任务时发出
AgentExecutionCompletedEvent：当 Agent 完成任务执行时发出
AgentExecutionErrorEvent：当 Agent 在执行过程中遇到错误时发出
LiteAgentExecutionStartedEvent：当 LiteAgent 开始执行时发出。包含 agent 信息、工具和消息。
LiteAgentExecutionCompletedEvent：当 LiteAgent 完成执行时发出。包含 agent 信息和输出。
LiteAgentExecutionErrorEvent：当 LiteAgent 在执行过程中遇到错误时发出。包含 agent 信息和错误消息。
AgentEvaluationStartedEvent：当 agent 评估开始时发出。包含 agent ID、agent 角色、可选 task ID 和迭代编号。
AgentEvaluationCompletedEvent：当 agent 评估完成时发出。包含 agent ID、agent 角色、可选 task ID、迭代编号、指标类别和分数。
AgentEvaluationFailedEvent：当 agent 评估失败时发出。包含 agent ID、agent 角色、可选 task ID、迭代编号和错误消息。

Task 事件

TaskStartedEvent：当 Task 开始执行时发出
TaskCompletedEvent：当 Task 完成执行时发出
TaskFailedEvent：当 Task 未能完成执行时发出
TaskEvaluationEvent：当 Task 被评估时发出

工具使用事件

ToolUsageStartedEvent：当工具执行开始时发出
ToolUsageFinishedEvent：当工具执行完成时发出
ToolUsageErrorEvent：当工具执行遇到错误时发出
ToolValidateInputErrorEvent：当工具输入验证遇到错误时发出
ToolExecutionErrorEvent：当工具执行遇到错误时发出
ToolSelectionErrorEvent：当选择工具时出现错误时发出

MCP 事件

MCPConnectionStartedEvent：当开始连接 MCP 服务器时发出。包含服务器名称、URL、传输类型、连接超时时间，以及是否为重连尝试。
MCPConnectionCompletedEvent：当成功连接到 MCP 服务器时发出。包含服务器名称、以毫秒计的连接时长，以及是否为重连。
MCPConnectionFailedEvent：当连接 MCP 服务器失败时发出。包含服务器名称、错误消息和错误类型（timeout、authentication、network 等）。
MCPToolExecutionStartedEvent：当开始执行 MCP 工具时发出。包含服务器名称、工具名称和工具参数。
MCPToolExecutionCompletedEvent：当 MCP 工具执行成功完成时发出。包含服务器名称、工具名称、结果和以毫秒计的执行时长。
MCPToolExecutionFailedEvent：当 MCP 工具执行失败时发出。包含服务器名称、工具名称、错误消息和错误类型（timeout、validation、server_error 等）。
MCPConfigFetchFailedEvent：当获取 MCP 服务器配置失败时发出（例如 MCP 未在你的账户中连接、API 错误，或获取配置后连接失败）。包含 slug、错误消息和错误类型（not_connected、api_error、connection_failed）。

知识事件

KnowledgeRetrievalStartedEvent：当开始知识检索时发出
KnowledgeRetrievalCompletedEvent：当知识检索完成时发出
KnowledgeQueryStartedEvent：当开始知识查询时发出
KnowledgeQueryCompletedEvent：当知识查询完成时发出
KnowledgeQueryFailedEvent：当知识查询失败时发出
KnowledgeSearchQueryFailedEvent：当知识搜索查询失败时发出

LLM Guardrail 事件

LLMGuardrailStartedEvent：当 guardrail 验证开始时发出。包含正在应用的 guardrail 详情和重试次数。
LLMGuardrailCompletedEvent：当 guardrail 验证完成时发出。包含验证成功/失败的详情、结果，以及如有的话错误消息。
LLMGuardrailFailedEvent：当 guardrail 验证失败时发出。包含错误消息和重试次数。

Flow 事件

FlowCreatedEvent：当 Flow 被创建时发出
FlowStartedEvent：当 Flow 开始执行时发出
FlowFinishedEvent：当 Flow 完成执行时发出
FlowPausedEvent：当 Flow 暂停等待人工反馈时发出。包含 flow 名称、flow ID、方法名、当前状态、请求反馈时显示的消息，以及用于路由的可选结果列表。
FlowPlotEvent：当 Flow 被绘图时发出
MethodExecutionStartedEvent：当 Flow 方法开始执行时发出
MethodExecutionFinishedEvent：当 Flow 方法完成执行时发出
MethodExecutionFailedEvent：当 Flow 方法无法完成执行时发出
MethodExecutionPausedEvent：当 Flow 方法暂停等待人工反馈时发出。包含 flow 名称、方法名、当前状态、flow ID、请求反馈时显示的消息，以及用于路由的可选结果列表。

人在回路事件

FlowInputRequestedEvent：当 Flow 通过 Flow.ask() 请求用户输入时发出。包含 flow 名称、方法名、向用户显示的问题或提示，以及可选元数据（例如用户 ID、频道、会话上下文）。
FlowInputReceivedEvent：当 Flow.ask() 之后收到用户输入时发出。包含 flow 名称、方法名、原始问题、用户响应（若超时则为 None）、可选请求元数据，以及来自提供商的可选响应元数据（例如是谁响应的、线程 ID、时间戳）。
HumanFeedbackRequestedEvent：当由 @human_feedback 装饰的方法需要人工审核者输入时发出。包含 flow 名称、方法名、展示给人工审核者的方法输出、请求反馈时显示的消息，以及用于路由的可选结果列表。
HumanFeedbackReceivedEvent：当人工对 @human_feedback 装饰的方法提供反馈时发出。包含 flow 名称、方法名、人工提供的原始文本反馈，以及折叠后的 outcome 字符串（如果指定了 emit）。

LLM 事件

LLMCallStartedEvent：当 LLM 调用开始时发出
LLMCallCompletedEvent：当 LLM 调用完成时发出
LLMCallFailedEvent：当 LLM 调用失败时发出
LLMStreamChunkEvent：在流式 LLM 响应接收每个 chunk 时发出
LLMThinkingChunkEvent：当从思考型模型接收到思考/推理 chunk 时发出。包含 chunk 文本和可选响应 ID。

记忆事件

MemoryQueryStartedEvent：当记忆查询开始时发出。包含查询、limit，以及可选的分数阈值。
MemoryQueryCompletedEvent：当记忆查询成功完成时发出。包含查询、结果、limit、分数阈值和查询执行时间。
MemoryQueryFailedEvent：当记忆查询失败时发出。包含查询、limit、分数阈值和错误消息。
MemorySaveStartedEvent：当记忆保存操作开始时发出。包含要保存的值、元数据，以及可选的 agent 角色。
MemorySaveCompletedEvent：当记忆保存操作成功完成时发出。包含已保存的值、元数据、agent 角色和保存执行时间。
MemorySaveFailedEvent：当记忆保存操作失败时发出。包含值、元数据、agent 角色和错误消息。
MemoryRetrievalStartedEvent：当任务提示词的记忆检索开始时发出。包含可选的 task ID。
MemoryRetrievalCompletedEvent：当任务提示词的记忆检索成功完成时发出。包含 task ID、记忆内容和检索执行时间。
MemoryRetrievalFailedEvent：当任务提示词的记忆检索失败时发出。包含可选的 task ID 和错误消息。

推理事件

AgentReasoningStartedEvent：当 agent 开始对任务进行推理时发出。包含 agent 角色、task ID 和尝试次数。
AgentReasoningCompletedEvent：当 agent 完成推理过程时发出。包含 agent 角色、task ID、生成的计划，以及 agent 是否准备继续。
AgentReasoningFailedEvent：当推理过程失败时发出。包含 agent 角色、task ID 和错误消息。

观察事件

StepObservationStartedEvent：当 Planner 开始观察某一步的结果时发出。它会在每次步骤执行后、观察 LLM 调用之前触发。包含 agent 角色、步骤编号和步骤描述。
StepObservationCompletedEvent：当 Planner 完成对某一步结果的观察时发出。包含该步骤是否成功完成、学到的关键信息、剩余计划是否仍然有效、是否需要全面重规划，以及建议的优化。
StepObservationFailedEvent：当观察用的 LLM 调用本身失败时发出。系统会默认继续执行计划。包含错误消息。
PlanRefinementEvent：当 Planner 在不进行全面重规划的情况下细化后续步骤描述时发出。包含被细化的步骤数量和所应用的细化内容。
PlanReplanTriggeredEvent：当 Planner 因为剩余计划被判定为根本错误而触发全面重规划时发出。包含重规划原因、重规划次数，以及保留的已完成步骤数。
GoalAchievedEarlyEvent：当 Planner 检测到目标已提前达成、后续步骤将被跳过时发出。包含剩余步骤数和已完成步骤数。

A2A（Agent-to-Agent）事件

委派事件

A2ADelegationStartedEvent：当 A2A 委派开始时发出。包含端点 URL、任务描述、agent ID、上下文 ID、是否多轮、轮次编号、agent 卡片元数据、协议版本、提供商信息，以及可选的技能 ID。
A2ADelegationCompletedEvent：当 A2A 委派完成时发出。包含完成状态（completed、input_required、failed 等）、结果、错误消息、上下文 ID 和 agent 卡片元数据。
A2AParallelDelegationStartedEvent：当并行委派给多个 A2A agent 开始时发出。包含端点列表和任务描述。
A2AParallelDelegationCompletedEvent：当并行委派给多个 A2A agent 完成时发出。包含端点列表、成功数量、失败数量和结果摘要。

对话事件

A2AConversationStartedEvent：在多轮 A2A 对话开始时、第一次消息交换之前发出一次。包含 agent ID、端点、上下文 ID、agent 卡片元数据、协议版本和提供商信息。
A2AMessageSentEvent：当消息发送给 A2A agent 时发出。包含消息内容、轮次编号、上下文 ID、消息 ID，以及是否为多轮。
A2AResponseReceivedEvent：当从 A2A agent 收到响应时发出。包含响应内容、轮次编号、上下文 ID、消息 ID、状态，以及是否为最终响应。
A2AConversationCompletedEvent：在多轮 A2A 对话结束时发出一次。包含最终状态（completed 或 failed）、最终结果、错误消息、上下文 ID 和总轮次数。

流式事件

A2AStreamingStartedEvent：当 A2A 委派进入流式模式时发出。包含 task ID、上下文 ID、端点、轮次编号，以及是否为多轮。
A2AStreamingChunkEvent：当接收到流式 chunk 时发出。包含 chunk 文本、chunk 索引、是否为最后一个 chunk、task ID、上下文 ID 和轮次编号。

轮询与推送通知事件

A2APollingStartedEvent：当 A2A 委派进入轮询模式时发出。包含 task ID、上下文 ID、轮询间隔（秒）和端点。
A2APollingStatusEvent：在每次轮询迭代时发出。包含 task ID、上下文 ID、当前任务状态、已过秒数和轮询次数。
A2APushNotificationRegisteredEvent：当注册推送通知回调时发出。包含 task ID、上下文 ID、回调 URL 和端点。
A2APushNotificationReceivedEvent：当从远程 A2A agent 收到推送通知时发出。包含 task ID、上下文 ID 和当前状态。
A2APushNotificationSentEvent：当向回调 URL 发送推送通知时发出。包含 task ID、上下文 ID、回调 URL、状态、是否发送成功，以及可选错误消息。
A2APushNotificationTimeoutEvent：当推送通知等待超时时发出。包含 task ID、上下文 ID 和超时时长（秒）。

连接与身份验证事件

A2AAgentCardFetchedEvent：当成功获取 agent card 时发出。包含端点、agent 名称、agent 卡片元数据、协议版本、提供商信息、是否为缓存结果，以及以毫秒计的获取时间。
A2AAuthenticationFailedEvent：当对 A2A agent 的身份验证失败时发出。包含端点、尝试的 auth 类型（例如 bearer、oauth2、api_key）、错误消息和 HTTP 状态码。
A2AConnectionErrorEvent：当 A2A 通信中发生连接错误时发出。包含端点、错误消息、错误类型（例如 timeout、connection_refused、dns_error）、HTTP 状态码和正在尝试的操作。
A2ATransportNegotiatedEvent：当与 A2A agent 协商传输协议时发出。包含协商后的传输方式、协商后的 URL、选择来源（client_preferred、server_preferred、fallback），以及客户端/服务器支持的传输方式。
A2AContentTypeNegotiatedEvent：当与 A2A agent 协商内容类型时发出。包含客户端/服务器输入输出模式、协商后的输入输出模式，以及协商是否成功。

产物事件

A2AArtifactReceivedEvent：当从远程 A2A agent 接收到产物时发出。包含 task ID、产物 ID、产物名称、描述、MIME 类型、字节大小，以及内容是否应追加。

服务器任务事件

A2AServerTaskStartedEvent：当 A2A 服务器任务执行开始时发出。包含 task ID 和上下文 ID。
A2AServerTaskCompletedEvent：当 A2A 服务器任务执行完成时发出。包含 task ID、上下文 ID 和结果。
A2AServerTaskCanceledEvent：当 A2A 服务器任务执行被取消时发出。包含 task ID 和上下文 ID。
A2AServerTaskFailedEvent：当 A2A 服务器任务执行失败时发出。包含 task ID、上下文 ID 和错误消息。

上下文生命周期事件

A2AContextCreatedEvent：当创建 A2A 上下文时发出。上下文会在对话或工作流中分组相关任务。包含上下文 ID 和创建时间戳。
A2AContextExpiredEvent：当 A2A 上下文因 TTL 过期时发出。包含上下文 ID、创建时间戳、存活秒数和任务数量。
A2AContextIdleEvent：当 A2A 上下文进入空闲状态（在配置阈值内没有活动）时发出。包含上下文 ID、空闲时长（秒）和任务数量。
A2AContextCompletedEvent：当 A2A 上下文中的所有任务完成时发出。包含上下文 ID、任务总数和持续时间（秒）。
A2AContextPrunedEvent：当 A2A 上下文被清理（删除）时发出。包含上下文 ID、任务数量和存活秒数。

事件处理器结构

每个事件处理器接收两个参数：

source：发出该事件的对象
event：事件实例，包含该事件特有的数据

事件对象的结构取决于事件类型，但所有事件都继承自 BaseEvent，并包含：

timestamp：事件发出的时间
type：事件类型的字符串标识符

其他字段会因事件类型而异。例如，CrewKickoffCompletedEvent 包含 crew_name 和 output 字段。

进阶用法：作用域处理器

对于临时性的事件处理（适用于测试或特定操作），你可以使用 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")

    # 执行会发出事件的操作

# 在上下文之外，临时处理器会被移除

使用场景

事件监听器可用于多种目的：

日志与监控：跟踪你的 Crew 执行并记录重要事件
分析：收集 Crew 的性能与行为数据
调试：设置临时监听器以排查特定问题
集成：将 CrewAI 连接到外部系统，如监控平台、数据库或通知服务
自定义行为：基于特定事件触发自定义动作

最佳实践

保持处理器轻量：事件处理器应尽量轻量，避免阻塞操作
错误处理：在事件处理器中加入适当的错误处理，防止异常影响主执行流程
清理资源：如果监听器分配了资源，请确保正确清理
选择性监听：只监听你实际需要处理的事件
测试：将事件监听器单独测试，确保其行为符合预期

利用 CrewAI 的事件系统，你可以扩展其功能，并将其无缝集成到现有基础设施中。