BytesRouter

主题

CrewAI 文档版本 跟随上游 源快照 9e213e22 更新于 2026-10-18

事件监听器

借助 CrewAI 事件构建自定义集成与监控

概述

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

工作原理

CrewAI 采用事件总线架构，在整个执行生命周期中发出事件。事件系统基于以下组件构建：

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

当 CrewAI 中发生特定操作时（例如团队开始执行、智能体完成任务或使用工具），系统会发出相应的事件。你可以为这些事件注册处理函数，以便在事件发生时执行自定义代码。

NOTE

CrewAI AMP 提供了内置的提示词追踪功能，利用事件系统跟踪、存储并可视化所有提示词、补全内容及相关元数据。这为你的智能体运行提供了强大的调试能力和透明度。

Prompt Tracing Dashboard

借助提示词追踪，你可以：

  • 查看发送给 LLM 的所有提示词的完整历史
  • 跟踪 token 用量与成本
  • 调试智能体推理失败
  • 与团队共享提示词序列
  • 比较不同的提示词策略
  • 导出追踪记录用于合规与审计

创建自定义事件监听器

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

  1. 创建一个继承自 BaseEventListener 的类
  2. 实现 setup_listeners 方法
  3. 为你关注的事件注册处理函数
  4. 在合适的文件中创建监听器实例

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

python
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 实现中导入并实例化

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

对于基于 Crew 的应用

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

python
# 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 的应用

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

python
# 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):
        # ...

这可以确保你的监听器在 Crew 或 Flow 执行时被加载并处于活跃状态。

方式 2：为你的监听器创建包

如需更结构化的方式，尤其是当你有多个监听器时：

  1. 为你的监听器创建一个包：
my_project/
  ├── listeners/
  │   ├── __init__.py
  │   ├── my_custom_listener.py
  │   └── another_listener.py
  1. 在 my_custom_listener.py 中，定义你的监听器类并创建实例：
python
# 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()
  1. 在 __init__.py 中导入监听器实例，确保它们被加载：
python
# __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']
  1. 在你的 Crew 或 Flow 文件中导入监听器包：
python
# In your crew.py or flow.py file
import my_project.listeners  # This loads all your listeners

class MyCustomCrew:
    # Your crew implementation...

第三方事件监听器正是这样在 CrewAI 代码库中注册的。

可用的事件类型

CrewAI 提供了丰富的事件供你监听：

团队事件

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

智能体事件

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

任务事件

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

工具使用事件

  • 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 护栏事件

  • LLMGuardrailStartedEvent：护栏校验开始时触发。包含所应用护栏的详细信息以及重试次数。
  • LLMGuardrailCompletedEvent：护栏校验完成时触发。包含校验成功/失败的详细信息、结果，以及错误消息（如有）。
  • LLMGuardrailFailedEvent：护栏校验失败时触发。包含错误消息和重试次数。

流程事件

  • FlowCreatedEvent：Flow 创建时触发
  • FlowStartedEvent：Flow 开始执行时触发
  • FlowFinishedEvent：Flow 完成执行时触发
  • FlowFailedEvent：Flow 执行失败时触发。包含流程名称以及导致执行终止的异常。
  • FlowPausedEvent：Flow 暂停等待人工反馈时触发。包含流程名称、流程 ID、方法名称、当前状态、请求反馈时展示的消息，以及用于路由的可能的输出结果的可选列表。
  • FlowPlotEvent：Flow 被绘制时触发
  • MethodExecutionStartedEvent：Flow 方法开始执行时触发
  • MethodExecutionFinishedEvent：Flow 方法完成执行时触发
  • MethodExecutionFailedEvent：Flow 方法未能完成执行时触发
  • MethodExecutionPausedEvent：Flow 方法暂停等待人工反馈时触发。包含流程名称、方法名称、当前状态、流程 ID、请求反馈时展示的消息，以及用于路由的可能的输出结果的可选列表。

人在回路事件

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

LLM 事件

  • LLMCallStartedEvent：LLM 调用开始时触发
  • LLMCallCompletedEvent：LLM 调用完成时触发
  • LLMCallFailedEvent：LLM 调用失败时触发
  • LLMStreamChunkEvent：在 LLM 响应流式传输过程中每收到一个分块时触发
  • LLMThinkingChunkEvent：从思考模型收到思考/推理分块时触发。包含分块文本和可选的响应 ID。

记忆事件

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

推理事件

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

观察事件

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

A2A（智能体到智能体）事件

委派事件

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

对话事件

  • A2AConversationStartedEvent：在多轮 A2A 对话开始时、首次消息交换之前触发一次。包含智能体 ID、端点、上下文 ID、智能体卡片元数据、协议版本和服务商信息。
  • A2AMessageSentEvent：向 A2A 智能体发送消息时触发。包含消息内容、轮次编号、上下文 ID、消息 ID，以及是否为多轮。
  • A2AResponseReceivedEvent：收到来自 A2A 智能体的响应时触发。包含响应内容、轮次编号、上下文 ID、消息 ID、状态，以及是否为最终响应。
  • A2AConversationCompletedEvent：在多轮 A2A 对话结束时触发一次。包含最终状态（completed 或 failed）、最终结果、错误消息、上下文 ID 和总轮次数。

流式传输事件

  • 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 智能体协商内容类型时发出。包含客户端/服务端的输入/输出模式、协商后的输入/输出模式，以及协商是否成功。

产物事件

  • 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：事件实例，包含特定于事件的数据

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

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

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

高级用法：作用域处理器

对于临时事件处理（适用于测试或特定操作），你可以使用 scoped_handlers 上下文管理器：

python
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. 测试：在隔离环境中测试事件监听器，以确保它们按预期运行

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

友情链接

  • Agents.md面向 AI 编码助手的仓库说明文件规范
  • LangChainLangChain 是一个开放的Agent平台
  • LangSmithAgent 工程平台：可观测与评估、部署、Fleet、Engine 等
  • EmbabelEmbabel 是一个JVM平台Agent框架
  • SpringAISpringAI 是一个基于 Spring 的 AI 框架
  • MCPModel Context Protocol，连接 AI 应用与外部数据/工具的开放标准
  • LangChain4jJava 生态的 LLM 应用开发框架
  • LlamaIndex把数据接入 LLM 的框架与文档平台
  • LlamaParse智能体式 OCR 文档解析：把 PDF、扫描件变成 LLM 可用的文本与结构化数据
  • LiteParse本地优先、轻量的文档解析库，不需要 API key
  • CrewAI多智能体编排框架：以 Crew/Flow 组织自主协作的 AI Agent
  • MAFMicrosoft Agent Framework，企业级 .NET/Python 智能体框架

BytesRouter 是社区翻译站点，所有译文仅供参考，一切以官方英文原文为准。

© 2026 BytesRouter · 译文采用 CC BY 4.0 许可