mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-09 16:18:30 +00:00
f845fac4da
- Renamed `CrewEvent` to `BaseEvent` across the codebase for consistency - Created a `CrewBaseEvent` that automatically identifies fingerprints for DRY - Added a new `to_json()` method for serializing events
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from typing import Any, Optional
|
|
|
|
from crewai.tasks.task_output import TaskOutput
|
|
from crewai.utilities.events.base_events import BaseEvent
|
|
|
|
|
|
class TaskStartedEvent(BaseEvent):
|
|
"""Event emitted when a task starts"""
|
|
|
|
type: str = "task_started"
|
|
context: Optional[str]
|
|
task: Optional[Any] = None
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the task
|
|
if hasattr(self.task, "fingerprint") and self.task.fingerprint:
|
|
self.source_fingerprint = self.task.fingerprint.uuid_str
|
|
self.source_type = "task"
|
|
if (
|
|
hasattr(self.task.fingerprint, "metadata")
|
|
and self.task.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.task.fingerprint.metadata
|
|
|
|
|
|
class TaskCompletedEvent(BaseEvent):
|
|
"""Event emitted when a task completes"""
|
|
|
|
output: TaskOutput
|
|
type: str = "task_completed"
|
|
task: Optional[Any] = None
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the task
|
|
if hasattr(self.task, "fingerprint") and self.task.fingerprint:
|
|
self.source_fingerprint = self.task.fingerprint.uuid_str
|
|
self.source_type = "task"
|
|
if (
|
|
hasattr(self.task.fingerprint, "metadata")
|
|
and self.task.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.task.fingerprint.metadata
|
|
|
|
|
|
class TaskFailedEvent(BaseEvent):
|
|
"""Event emitted when a task fails"""
|
|
|
|
error: str
|
|
type: str = "task_failed"
|
|
task: Optional[Any] = None
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the task
|
|
if hasattr(self.task, "fingerprint") and self.task.fingerprint:
|
|
self.source_fingerprint = self.task.fingerprint.uuid_str
|
|
self.source_type = "task"
|
|
if (
|
|
hasattr(self.task.fingerprint, "metadata")
|
|
and self.task.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.task.fingerprint.metadata
|
|
|
|
|
|
class TaskEvaluationEvent(BaseEvent):
|
|
"""Event emitted when a task evaluation is completed"""
|
|
|
|
type: str = "task_evaluation"
|
|
evaluation_type: str
|
|
task: Optional[Any] = None
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the task
|
|
if hasattr(self.task, "fingerprint") and self.task.fingerprint:
|
|
self.source_fingerprint = self.task.fingerprint.uuid_str
|
|
self.source_type = "task"
|
|
if (
|
|
hasattr(self.task.fingerprint, "metadata")
|
|
and self.task.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.task.fingerprint.metadata
|