事件监听器

文档索引

获取完整文档索引： https://docs.crewai.com.cn/llms.txt

在深入了解之前，请使用此文件来浏览所有可用页面。

​

概述

​

工作原理

1. CrewAIEventsBus：一个单例事件总线，用于管理事件的注册和发布。
2. BaseEvent：系统中所有事件的基类。
3. BaseEventListener：用于创建自定义事件监听器的抽象基类。

CrewAI AMP 提供了一项内置的提示追踪（Prompt Tracing）功能，该功能利用事件系统来跟踪、存储和可视化所有提示（prompts）、补全（completions）及相关的元数据。这为您的代理操作提供了强大的调试能力和透明度。通过提示追踪，您可以：

• 查看发送给 LLM 的所有提示的完整历史记录
• 跟踪令牌使用情况和成本
• 调试代理推理失败的原因
• 与团队共享提示序列
• 比较不同的提示策略
• 导出追踪记录以进行合规和审计

​

创建自定义事件监听器

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）的应用程序

# 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=[...],
            # ...
        )

​

对于基于流程（Flow）的应用程序

# In your main.py or flow.py file
from crewai.flow import Flow, listen, start
from my_listeners import MyCustomListener

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

class MyCustomFlow(Flow):
    # Your flow implementation...

    @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 ...

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

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

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

# Optionally export them if you need to access them elsewhere
__all__ = ['my_custom_listener', 'another_listener']

4. 在您的团队或流程文件中导入您的监听器包

# In your crew.py or flow.py file
import my_project.listeners  # This loads all your listeners

class MyCustomCrew:
    # Your crew implementation...

​

可用事件类型

​

团队（Crew）事件

• CrewKickoffStartedEvent：当团队开始执行时发布
• CrewKickoffCompletedEvent：当团队完成执行时发布
• CrewKickoffFailedEvent：当团队执行失败时发布
• CrewTestStartedEvent：当团队开始测试时发布
• CrewTestCompletedEvent：当团队完成测试时发布
• CrewTestFailedEvent：当团队测试失败时发布
• CrewTrainStartedEvent：当团队开始训练时发布
• CrewTrainCompletedEvent：当团队完成训练时发布
• CrewTrainFailedEvent：当团队训练失败时发布
• CrewTestResultEvent：当团队测试结果可用时发布。包含质量分数、执行持续时间和所使用的模型。

​

代理（Agent）事件

• AgentExecutionStartedEvent：当代理开始执行任务时发布
• AgentExecutionCompletedEvent：当代理完成任务执行时发布
• AgentExecutionErrorEvent：当代理在执行期间遇到错误时发布
• LiteAgentExecutionStartedEvent：当轻量级代理开始执行时发布。包含代理信息、工具和消息。
• LiteAgentExecutionCompletedEvent：当轻量级代理完成执行时发布。包含代理信息和输出。
• LiteAgentExecutionErrorEvent：当轻量级代理在执行期间遇到错误时发布。包含代理信息和错误消息。
• AgentEvaluationStartedEvent：当代理评估开始时发布。包含代理 ID、代理角色、可选任务 ID 和迭代次数。
• AgentEvaluationCompletedEvent：当代理评估完成时发布。包含代理 ID、代理角色、可选任务 ID、迭代次数、指标类别和分数。
• AgentEvaluationFailedEvent：当代理评估失败时发布。包含代理 ID、代理角色、可选任务 ID、迭代次数和错误消息。

​

任务（Task）事件

• TaskStartedEvent：当任务开始执行时发布
• TaskCompletedEvent：当任务完成执行时发布
• TaskFailedEvent：当任务执行失败时发布
• TaskEvaluationEvent：当任务被评估时发布

​

工具使用（Tool Usage）事件

• 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）。

​

知识（Knowledge）事件

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

​

LLM 防护栏（Guardrail）事件

• LLMGuardrailStartedEvent：当防护栏验证开始时发布。包含有关所应用防护栏的信息和重试计数。
• LLMGuardrailCompletedEvent：当防护栏验证完成时发布。包含有关验证成功/失败的信息、结果以及任何错误消息。
• LLMGuardrailFailedEvent：当防护栏验证失败时发布。包含错误消息和重试计数。

​

流程（Flow）事件

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

​

人机协作（Human In The Loop）事件

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

​

LLM 事件

• LLMCallStartedEvent：当 LLM 调用开始时发布
• LLMCallCompletedEvent：当 LLM 调用完成时发布
• LLMCallFailedEvent：当 LLM 调用失败时发布
• LLMStreamChunkEvent：在流式传输 LLM 响应期间，接收到每个数据块时发布
• LLMThinkingChunkEvent：当从思维模型（thinking model）接收到思维/推理数据块时发布。包含块文本和可选的响应 ID。

​

记忆（Memory）事件

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

​

推理（Reasoning）事件

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

​

观察（Observation）事件

• StepObservationStartedEvent：当规划器开始观察步骤结果时发布。在每次步骤执行后、观察 LLM 调用前触发。包含代理角色、步骤编号和步骤描述。
• StepObservationCompletedEvent：当规划器完成观察步骤结果时发布。包含步骤是否成功完成、学到的关键信息、剩余计划是否依然有效、是否需要完全重新规划以及建议的改进。
• StepObservationFailedEvent：当观察 LLM 调用本身失败时发布。系统默认继续原计划。包含错误消息。
• PlanRefinementEvent：当规划器在无需完全重新规划的情况下优化后续步骤描述时发布。包含已优化的步骤数量和所应用的优化措施。
• PlanReplanTriggeredEvent：当规划器认为原剩余计划根本错误而触发完全重新规划时发布。包含重新规划原因、重新规划计数和已保留的已完成步骤数量。
• GoalAchievedEarlyEvent：当规划器检测到目标已提前达成，从而跳过剩余步骤时发布。包含剩余步骤数量和已完成步骤数量。

​

A2A（代理间通信）事件

​

委派（Delegation）事件

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

​

对话（Conversation）事件

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

​

流式传输（Streaming）事件

• A2AStreamingStartedEvent：当 A2A 委派进入流式传输模式时发布。包含任务 ID、上下文 ID、端点、轮次编号以及是否为多轮对话。
• A2AStreamingChunkEvent：当收到流式传输数据块时发布。包含数据块文本、数据块索引、是否为最后一块、任务 ID、上下文 ID 和轮次编号。

​

轮询与推送通知事件

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

​

连接与认证事件

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

​

工件（Artifact）事件

• A2AArtifactReceivedEvent：当从远程 A2A 代理收到工件时发布。包含任务 ID、工件 ID、名称、描述、MIME 类型、大小（字节）以及是否应追加内容。

​

服务器任务事件

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

​

上下文生命周期事件

• A2AContextCreatedEvent：当 A2A 上下文被创建时发布。上下文将对话或工作流中的相关任务归组在一起。包含上下文 ID 和创建时间戳。
• A2AContextExpiredEvent：当 A2A 上下文因生存时间（TTL）过期时发布。包含上下文 ID、创建时间戳、存活时长（秒）和任务计数。
• A2AContextIdleEvent：当 A2A 上下文变为空闲时（在配置的阈值内无活动）发布。包含上下文 ID、空闲时长（秒）和任务计数。
• A2AContextCompletedEvent：当 A2A 上下文中的所有任务完成时发布。包含上下文 ID、总任务数和持续时间（秒）。
• A2AContextPrunedEvent：当 A2A 上下文被修剪（删除）时发布。包含上下文 ID、任务计数和存活时长（秒）。

​

事件处理程序结构

1. source：发布事件的对象
2. event：事件实例，包含特定于事件的数据

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

​

高级用法：作用域处理程序

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

# Outside the context, the temporary handler is removed

​

用例

1. 日志和监控：跟踪团队执行情况并记录重要事件
2. 分析：收集有关团队性能和行为的数据
3. 调试：设置临时监听器以调试特定问题
4. 集成：将 CrewAI 与监控平台、数据库或通知服务等外部系统连接
5. 自定义行为：根据特定事件触发自定义操作

​

最佳实践

1. 保持轻量：事件处理程序应保持轻量，并避免阻塞操作
2. 错误处理：在事件处理程序中包含适当的错误处理，防止异常影响主执行流程
3. 清理：如果监听器分配了资源，请确保它们得到正确清理
4. 选择性监听：仅监听您确实需要处理的事件
5. 测试：独立测试您的事件监听器，确保其表现符合预期