mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-08 23:58:34 +00:00
a5b5c8ab37
* fix: possible fix for Thinking stuck * feat: add agent logging events for execution tracking - Introduced AgentLogsStartedEvent and AgentLogsExecutionEvent to enhance logging capabilities during agent execution. - Updated CrewAgentExecutor to emit these events at the start and during execution, respectively. - Modified EventListener to handle the new logging events and format output accordingly in the console. - Enhanced ConsoleFormatter to display agent logs in a structured format, improving visibility of agent actions and outputs. * drop emoji * refactor: improve code structure and logging in LiteAgent and ConsoleFormatter - Refactored imports in lite_agent.py for better readability. - Enhanced guardrail property initialization in LiteAgent. - Updated logging functionality to emit AgentLogsExecutionEvent for better tracking. - Modified ConsoleFormatter to include tool arguments and final output in status updates. - Improved output formatting for long text in ConsoleFormatter. * fix tests --------- Co-authored-by: Eduardo Chiarotti <dudumelgaco@hotmail.com>
126 lines
3.9 KiB
Python
126 lines
3.9 KiB
Python
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
|
|
|
|
from crewai.agents.agent_builder.base_agent import BaseAgent
|
|
from crewai.tools.base_tool import BaseTool
|
|
from crewai.tools.structured_tool import CrewStructuredTool
|
|
|
|
from .base_events import BaseEvent
|
|
|
|
if TYPE_CHECKING:
|
|
from crewai.agents.agent_builder.base_agent import BaseAgent
|
|
|
|
|
|
class AgentExecutionStartedEvent(BaseEvent):
|
|
"""Event emitted when an agent starts executing a task"""
|
|
|
|
agent: BaseAgent
|
|
task: Any
|
|
tools: Optional[Sequence[Union[BaseTool, CrewStructuredTool]]]
|
|
task_prompt: str
|
|
type: str = "agent_execution_started"
|
|
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the agent
|
|
if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
|
|
self.source_fingerprint = self.agent.fingerprint.uuid_str
|
|
self.source_type = "agent"
|
|
if (
|
|
hasattr(self.agent.fingerprint, "metadata")
|
|
and self.agent.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.agent.fingerprint.metadata
|
|
|
|
|
|
class AgentExecutionCompletedEvent(BaseEvent):
|
|
"""Event emitted when an agent completes executing a task"""
|
|
|
|
agent: BaseAgent
|
|
task: Any
|
|
output: str
|
|
type: str = "agent_execution_completed"
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the agent
|
|
if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
|
|
self.source_fingerprint = self.agent.fingerprint.uuid_str
|
|
self.source_type = "agent"
|
|
if (
|
|
hasattr(self.agent.fingerprint, "metadata")
|
|
and self.agent.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.agent.fingerprint.metadata
|
|
|
|
|
|
class AgentExecutionErrorEvent(BaseEvent):
|
|
"""Event emitted when an agent encounters an error during execution"""
|
|
|
|
agent: BaseAgent
|
|
task: Any
|
|
error: str
|
|
type: str = "agent_execution_error"
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
# Set fingerprint data from the agent
|
|
if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
|
|
self.source_fingerprint = self.agent.fingerprint.uuid_str
|
|
self.source_type = "agent"
|
|
if (
|
|
hasattr(self.agent.fingerprint, "metadata")
|
|
and self.agent.fingerprint.metadata
|
|
):
|
|
self.fingerprint_metadata = self.agent.fingerprint.metadata
|
|
|
|
|
|
# New event classes for LiteAgent
|
|
class LiteAgentExecutionStartedEvent(BaseEvent):
|
|
"""Event emitted when a LiteAgent starts executing"""
|
|
|
|
agent_info: Dict[str, Any]
|
|
tools: Optional[Sequence[Union[BaseTool, CrewStructuredTool]]]
|
|
messages: Union[str, List[Dict[str, str]]]
|
|
type: str = "lite_agent_execution_started"
|
|
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
|
|
|
|
class LiteAgentExecutionCompletedEvent(BaseEvent):
|
|
"""Event emitted when a LiteAgent completes execution"""
|
|
|
|
agent_info: Dict[str, Any]
|
|
output: str
|
|
type: str = "lite_agent_execution_completed"
|
|
|
|
|
|
class LiteAgentExecutionErrorEvent(BaseEvent):
|
|
"""Event emitted when a LiteAgent encounters an error during execution"""
|
|
|
|
agent_info: Dict[str, Any]
|
|
error: str
|
|
type: str = "lite_agent_execution_error"
|
|
|
|
|
|
# New logging events
|
|
class AgentLogsStartedEvent(BaseEvent):
|
|
"""Event emitted when agent logs should be shown at start"""
|
|
|
|
agent_role: str
|
|
task_description: Optional[str] = None
|
|
verbose: bool = False
|
|
type: str = "agent_logs_started"
|
|
|
|
|
|
class AgentLogsExecutionEvent(BaseEvent):
|
|
"""Event emitted when agent logs should be shown during execution"""
|
|
|
|
agent_role: str
|
|
formatted_answer: Any
|
|
verbose: bool = False
|
|
type: str = "agent_logs_execution"
|
|
|
|
model_config = {"arbitrary_types_allowed": True}
|