andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-28 09:38:17 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Add verbose flag to EventListener for controlled logging

Browse Source
This commit is contained in:
Lorenze Jay
2025-03-11 08:43:56 -07:00
parent f2d53350d8
commit 7c46dce2f7
3 changed files with 365 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/crewai/crew.py
View File

@@ -248,7 +248,10 @@ class Crew(BaseModel):
@model_validator(mode="after") @model_validator(mode="after")
def set_private_attrs(self) -> "Crew": def set_private_attrs(self) -> "Crew":
"""Set private attributes.""" """Set private attributes."""
from crewai.utilities.events.event_listener import EventListener
self._cache_handler = CacheHandler() self._cache_handler = CacheHandler()
EventListener().verbose = self.verbose
self._logger = Logger(verbose=self.verbose) self._logger = Logger(verbose=self.verbose)
if self.output_log_file: if self.output_log_file:
self._file_handler = FileHandler(self.output_log_file) self._file_handler = FileHandler(self.output_log_file)

2
src/crewai/utilities/events/base_event_listener.py
View File

@@ -5,6 +5,8 @@ from crewai.utilities.events.crewai_event_bus import CrewAIEventsBus, crewai_eve
class BaseEventListener(ABC): class BaseEventListener(ABC):
verbose: bool = False
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.setup_listeners(crewai_event_bus) self.setup_listeners(crewai_event_bus)

698
src/crewai/utilities/events/event_listener.py
View File

@@ -1,6 +1,6 @@
from datetime import datetime from datetime import datetime
from io import StringIO from io import StringIO
from typing import Any, Dict from typing import Any, Dict, Optional
from pydantic import Field, PrivateAttr from pydantic import Field, PrivateAttr
from rich.console import Console from rich.console import Console
@@ -112,7 +112,7 @@ class EventListener(BaseEventListener):
prefix: str, prefix: str,
name: str, name: str,
style: str = "blue", style: str = "blue",
status: str = None, status: Optional[str] = None,
) -> None: ) -> None:
"""Update tree label with consistent formatting.""" """Update tree label with consistent formatting."""
label = Text() label = Text()
@@ -130,19 +130,23 @@ class EventListener(BaseEventListener):
# ----------- METHODS ----------- # ----------- METHODS -----------
def on_crew_start(self, source: Any, event: Any) -> None: def on_crew_start(self, source: Any, event: Any) -> None:
# Create the crew tree that will hold tasks if self.verbose:
self.current_crew_tree = Tree( self.current_crew_tree = Tree(
Text("🚀 Crew: ", style="cyan bold") + Text(event.crew_name, style="cyan") Text("🚀 Crew: ", style="cyan bold")
) + Text(event.crew_name, style="cyan")
)
content = Text() content = self._create_status_content(
content.append(f"NAME: {event.crew_name}", style="white") "Crew Execution Started",
content.append(f"\nID: {source.id}", style="blue") event.crew_name,
"cyan",
ID=source.id,
)
panel = self._create_panel(content, "Crew Execution Started", "cyan") panel = self._create_panel(content, "Crew Execution Started", "cyan")
self.console.print(panel)
self.console.print()
self.console.print(panel)
self.console.print() # Add spacing
self._telemetry.crew_execution_span(source, event.inputs) self._telemetry.crew_execution_span(source, event.inputs)
# ----------- CREW EVENTS ----------- # ----------- CREW EVENTS -----------
@@ -157,70 +161,67 @@ class EventListener(BaseEventListener):
# Handle telemetry # Handle telemetry
final_string_output = event.output.raw final_string_output = event.output.raw
self._telemetry.end_crew(source, final_string_output) self._telemetry.end_crew(source, final_string_output)
if self.verbose:
if self.current_crew_tree:
self._update_tree_label(
self.current_crew_tree,
"✅ Crew:",
event.crew_name or "Crew",
"green",
)
if self.current_crew_tree: completion_content = self._create_status_content(
# Update crew tree label to show completion "Crew Execution Completed",
crew_content = Text() event.crew_name or "Crew",
crew_content.append("✅ Crew: ", style="green bold") "green",
crew_content.append(event.crew_name or "Crew", style="green") ID=source.id,
self.current_crew_tree.label = crew_content )
# Create completion panel self.console.print(self.current_crew_tree)
completion_content = Text() self.console.print()
completion_content.append( panel = self._create_panel(
"Crew Execution Completed\n", style="green bold" completion_content, "Crew Completion", "green"
) )
completion_content.append("Name: ", style="white") self.console.print(panel)
completion_content.append(f"{event.crew_name}\n", style="green") self.console.print()
completion_content.append("ID: ", style="white")
completion_content.append(str(source.id), style="blue")
# Show final tree and completion panel
self.console.print(self.current_crew_tree)
self.console.print()
panel = self._create_panel(
completion_content, "Crew Completion", "green"
)
self.console.print(panel)
self.console.print()
@crewai_event_bus.on(CrewKickoffFailedEvent) @crewai_event_bus.on(CrewKickoffFailedEvent)
def on_crew_failed(source, event: CrewKickoffFailedEvent): def on_crew_failed(source, event: CrewKickoffFailedEvent):
if self.current_crew_tree: if self.verbose:
# Update crew tree label to show failure if self.current_crew_tree:
crew_content = Text() # Update crew tree label to show failure
crew_content.append("❌ Crew: ", style="red bold") crew_content = Text()
crew_content.append(event.crew_name or "Crew", style="red") crew_content.append("❌ Crew: ", style="red bold")
self.current_crew_tree.label = crew_content crew_content.append(event.crew_name or "Crew", style="red")
self.current_crew_tree.label = crew_content
# Create failure panel # Create failure panel
failure_content = Text() failure_content = Text()
failure_content.append("Crew Execution Failed\n", style="red bold") failure_content.append("Crew Execution Failed\n", style="red bold")
failure_content.append("Name: ", style="white") failure_content.append("Name: ", style="white")
failure_content.append(f"{event.crew_name}\n", style="red") failure_content.append(f"{event.crew_name}\n", style="red")
failure_content.append("ID: ", style="white") failure_content.append("ID: ", style="white")
failure_content.append(str(source.id), style="blue") failure_content.append(str(source.id), style="blue")
# Show final tree and failure panel # Show final tree and failure panel
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
self.console.print() self.console.print()
panel = self._create_panel(failure_content, "Crew Failure", "red") panel = self._create_panel(failure_content, "Crew Failure", "red")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
@crewai_event_bus.on(CrewTestFailedEvent) @crewai_event_bus.on(CrewTestFailedEvent)
def on_crew_test_failed(source, event: CrewTestFailedEvent): def on_crew_test_failed(source, event: CrewTestFailedEvent):
# Create test failure content if self.verbose:
failure_content = Text() failure_content = Text()
failure_content.append("❌ Crew Test Failed\n", style="red bold") failure_content.append("❌ Crew Test Failed\n", style="red bold")
failure_content.append("Crew: ", style="white") failure_content.append("Crew: ", style="white")
failure_content.append(event.crew_name or "Crew", style="red") failure_content.append(event.crew_name or "Crew", style="red")
panel = self._create_panel(failure_content, "Test Failure", "red") panel = self._create_panel(failure_content, "Test Failure", "red")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
@crewai_event_bus.on(CrewTrainStartedEvent) @crewai_event_bus.on(CrewTrainStartedEvent)
def on_crew_train_started(source, event: CrewTrainStartedEvent): def on_crew_train_started(source, event: CrewTrainStartedEvent):
@@ -238,15 +239,15 @@ class EventListener(BaseEventListener):
@crewai_event_bus.on(CrewTrainFailedEvent) @crewai_event_bus.on(CrewTrainFailedEvent)
def on_crew_train_failed(source, event: CrewTrainFailedEvent): def on_crew_train_failed(source, event: CrewTrainFailedEvent):
# Create training failure content if self.verbose:
failure_content = Text() failure_content = Text()
failure_content.append("❌ Crew Training Failed\n", style="red bold") failure_content.append("❌ Crew Training Failed\n", style="red bold")
failure_content.append("Crew: ", style="white") failure_content.append("Crew: ", style="white")
failure_content.append(event.crew_name or "Crew", style="red") failure_content.append(event.crew_name or "Crew", style="red")
panel = self._create_panel(failure_content, "Training Failure", "red") panel = self._create_panel(failure_content, "Training Failure", "red")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
# ----------- TASK EVENTS ----------- # ----------- TASK EVENTS -----------
@@ -255,23 +256,23 @@ class EventListener(BaseEventListener):
span = self._telemetry.task_started(crew=source.agent.crew, task=source) span = self._telemetry.task_started(crew=source.agent.crew, task=source)
self.execution_spans[source] = span self.execution_spans[source] = span
# Create task content if self.verbose:
task_content = Text() task_content = Text()
task_content.append("📋 Task: ", style="yellow bold") task_content.append("📋 Task: ", style="yellow bold")
task_content.append("\n Assigned to: ", style="white") task_content.append("\n Assigned to: ", style="white")
task_content.append(source.agent.role, style="green") task_content.append(source.agent.role, style="green")
task_content.append("\n Status: ", style="white") task_content.append("\n Status: ", style="white")
task_content.append("Waiting for agent...", style="yellow dim") task_content.append("Waiting for agent...", style="yellow dim")
# Add task to the crew tree # Add task to the crew tree
if self.current_crew_tree: if self.current_crew_tree:
self.current_task_branch = self.current_crew_tree.add(task_content) self.current_task_branch = self.current_crew_tree.add(task_content)
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
else: else:
panel = self._create_panel(task_content, "Task Started", "yellow") panel = self._create_panel(task_content, "Task Started", "yellow")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
@crewai_event_bus.on(TaskCompletedEvent) @crewai_event_bus.on(TaskCompletedEvent)
def on_task_completed(source, event: TaskCompletedEvent): def on_task_completed(source, event: TaskCompletedEvent):
@@ -281,27 +282,29 @@ class EventListener(BaseEventListener):
self._telemetry.task_ended(span, source, source.agent.crew) self._telemetry.task_ended(span, source, source.agent.crew)
self.execution_spans[source] = None self.execution_spans[source] = None
# Create completion content if self.verbose:
completion_content = Text() completion_content = Text()
completion_content.append("✅ Task Completed\n", style="green bold") completion_content.append("✅ Task Completed\n", style="green bold")
completion_content.append(f"Task: {str(source.id)}", style="white") completion_content.append(f"Task: {str(source.id)}", style="white")
completion_content.append("\nAgent: ", style="white") completion_content.append("\nAgent: ", style="white")
completion_content.append(source.agent.role, style="green") completion_content.append(source.agent.role, style="green")
# Update the tree if it exists # Update the tree if it exists
if self.current_crew_tree: if self.current_crew_tree:
# Find the task branch and update it with completion status for branch in self.current_crew_tree.children:
for branch in self.current_crew_tree.children: if source.description in branch.label:
if source.description in branch.label: branch.label = (
branch.label = Text("✅ ", style="green bold") + branch.label Text("✅ ", style="green bold") + branch.label
self.console.print(self.current_crew_tree) )
break self.console.print(self.current_crew_tree)
break
# Always show completion panel panel = self._create_panel(
panel = self._create_panel(completion_content, "Task Completion", "green") completion_content, "Task Completion", "green"
)
self.console.print(panel) self.console.print(panel)
self.console.print() # Add spacing self.console.print()
@crewai_event_bus.on(TaskFailedEvent) @crewai_event_bus.on(TaskFailedEvent)
def on_task_failed(source, event: TaskFailedEvent): def on_task_failed(source, event: TaskFailedEvent):
@@ -311,60 +314,62 @@ class EventListener(BaseEventListener):
self._telemetry.task_ended(span, source, source.agent.crew) self._telemetry.task_ended(span, source, source.agent.crew)
self.execution_spans[source] = None self.execution_spans[source] = None
# Create failure content if self.verbose:
failure_content = Text() failure_content = Text()
failure_content.append("❌ Task Failed\n", style="red bold") failure_content.append("❌ Task Failed\n", style="red bold")
failure_content.append("Task: ", style="white") failure_content.append("Task: ", style="white")
failure_content.append(source.description, style="red") failure_content.append(source.description, style="red")
if source.agent: if source.agent:
failure_content.append("\nAgent: ", style="white") failure_content.append("\nAgent: ", style="white")
failure_content.append(source.agent.role, style="red") failure_content.append(source.agent.role, style="red")
# Update the tree if it exists # Update the tree if it exists
if self.current_crew_tree: if self.current_crew_tree:
# Find the task branch and update it with failure status # Find the task branch and update it with failure status
for branch in self.current_crew_tree.children: for branch in self.current_crew_tree.children:
if source.description in branch.label: if source.description in branch.label:
branch.label = Text("❌ ", style="red bold") + branch.label branch.label = Text("❌ ", style="red bold") + branch.label
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
break break
# Show failure panel # Show failure panel
panel = self._create_panel(failure_content, "Task Failure", "red") panel = self._create_panel(failure_content, "Task Failure", "red")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
# ----------- AGENT EVENTS ----------- # ----------- AGENT EVENTS -----------
@crewai_event_bus.on(AgentExecutionStartedEvent) @crewai_event_bus.on(AgentExecutionStartedEvent)
def on_agent_execution_started(source, event: AgentExecutionStartedEvent): def on_agent_execution_started(source, event: AgentExecutionStartedEvent):
if self.current_task_branch: if self.verbose:
# Create agent execution branch if self.current_task_branch:
agent_content = Text() # Create agent execution branch with empty label
agent_content.append("🤖 Agent: ", style="green bold") self.current_agent_branch = self.current_task_branch.add("")
agent_content.append(event.agent.role, style="green") self._update_tree_label(
agent_content.append("\n Status: ", style="white") self.current_agent_branch,
agent_content.append("In Progress", style="blue bold") "🤖 Agent:",
event.agent.role,
"green",
"In Progress",
)
# Create a branch for the agent's activities self.console.print(self.current_crew_tree)
self.current_agent_branch = self.current_task_branch.add(agent_content) self.console.print()
self.console.print(self.current_crew_tree)
self.console.print()
@crewai_event_bus.on(AgentExecutionCompletedEvent) @crewai_event_bus.on(AgentExecutionCompletedEvent)
def on_agent_execution_completed(source, event: AgentExecutionCompletedEvent): def on_agent_execution_completed(source, event: AgentExecutionCompletedEvent):
if self.current_agent_branch: if self.verbose:
# Update agent branch to show completion if self.current_agent_branch:
agent_content = Text() self._update_tree_label(
agent_content.append("🤖 Agent: ", style="green bold") self.current_agent_branch,
agent_content.append(event.agent.role, style="green") "🤖 Agent:",
agent_content.append("\n Status: ", style="white") event.agent.role,
agent_content.append("✅ Completed", style="green bold") "green",
"✅ Completed",
)
# Update the agent branch label self.console.print(self.current_crew_tree)
self.current_agent_branch.label = agent_content self.console.print()
self.console.print(self.current_crew_tree)
self.console.print()
# ----------- FLOW EVENTS ----------- # ----------- FLOW EVENTS -----------
@@ -372,33 +377,32 @@ class EventListener(BaseEventListener):
def on_flow_created(source, event: FlowCreatedEvent): def on_flow_created(source, event: FlowCreatedEvent):
self._telemetry.flow_creation_span(event.flow_name) self._telemetry.flow_creation_span(event.flow_name)
# Create flow content for panel if self.verbose:
content = Text() content = self._create_status_content(
content.append("🌊 Starting Flow Execution\n\n", style="blue bold") "Starting Flow Execution", event.flow_name, "blue"
content.append("Name: ", style="white") )
content.append(event.flow_name, style="blue")
panel = self._create_panel(content, "Flow Execution", "blue") panel = self._create_panel(content, "Flow Execution", "blue")
self.console.print() self.console.print()
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
# Create and display the initial tree # Create and display the initial tree
flow_label = Text() flow_label = Text()
flow_label.append("🌊 Flow: ", style="blue bold") flow_label.append("🌊 Flow: ", style="blue bold")
flow_label.append(event.flow_name, style="blue") flow_label.append(event.flow_name, style="blue")
self.current_flow_tree = Tree(flow_label) self.current_flow_tree = Tree(flow_label)
# Add both creation steps to show progression # Add both creation steps to show progression
self.current_flow_tree.add(Text("✨ Created", style="blue")) self.current_flow_tree.add(Text("✨ Created", style="blue"))
self.current_flow_tree.add( self.current_flow_tree.add(
Text("✅ Initialization Complete", style="green") Text("✅ Initialization Complete", style="green")
) )
self.console.print(self.current_flow_tree) self.console.print(self.current_flow_tree)
self.console.print() self.console.print()
@crewai_event_bus.on(FlowStartedEvent) @crewai_event_bus.on(FlowStartedEvent)
def on_flow_started(source, event: FlowStartedEvent): def on_flow_started(source, event: FlowStartedEvent):
@@ -406,174 +410,187 @@ class EventListener(BaseEventListener):
event.flow_name, list(source._methods.keys()) event.flow_name, list(source._methods.keys())
) )
# Create flow tree with status if self.verbose:
flow_label = Text() self.current_flow_tree = Tree("")
flow_label.append("🌊 Flow: ", style="blue bold") self._update_tree_label(
flow_label.append(event.flow_name, style="blue") self.current_flow_tree,
flow_label.append("\n Status: ", style="white") "🌊 Flow:",
flow_label.append("In Progress", style="yellow") event.flow_name,
"blue",
"In Progress",
)
self.current_flow_tree = Tree(flow_label) # Add initial thinking state
self.current_flow_tree.add(Text("🧠 Initializing...", style="yellow"))
# Add initial thinking state self.console.print()
self.current_flow_tree.add(Text("🧠 Initializing...", style="yellow")) self.console.print(self.current_flow_tree)
self.console.print()
self.console.print()
self.console.print(self.current_flow_tree)
self.console.print()
@crewai_event_bus.on(FlowFinishedEvent) @crewai_event_bus.on(FlowFinishedEvent)
def on_flow_finished(source, event: FlowFinishedEvent): def on_flow_finished(source, event: FlowFinishedEvent):
if self.current_flow_tree: if self.verbose:
# Update flow tree label to show completion if self.current_flow_tree:
self.current_flow_tree.label = Text( self._update_tree_label(
"✅ Flow Finished: ", style="green bold" self.current_flow_tree,
) + Text(event.flow_name, style="green") "✅ Flow Finished:",
event.flow_name,
"green",
)
# Create completion content content = self._create_status_content(
content = Text() "Flow Execution Completed",
content.append("Flow Execution Completed\n", style="green bold") event.flow_name,
content.append("Name: ", style="white") "green",
content.append(f"{event.flow_name}\n", style="green") ID=source.flow_id,
content.append("ID: ", style="white") )
content.append(source.flow_id, style="blue")
panel = self._create_panel(content, "Flow Completion", "green") panel = self._create_panel(content, "Flow Completion", "green")
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
@crewai_event_bus.on(MethodExecutionStartedEvent) @crewai_event_bus.on(MethodExecutionStartedEvent)
def on_method_execution_started(source, event: MethodExecutionStartedEvent): def on_method_execution_started(source, event: MethodExecutionStartedEvent):
if self.current_flow_tree: if self.verbose:
# Find and update the method branch if self.current_flow_tree:
for branch in self.current_flow_tree.children: # Find and update the method branch
if event.method_name in branch.label: for branch in self.current_flow_tree.children:
self.current_method_branch = branch if event.method_name in branch.label:
branch.label = Text("🔄 Running: ", style="yellow bold") + Text( self.current_method_branch = branch
event.method_name, style="yellow" branch.label = Text(
) "🔄 Running: ", style="yellow bold"
break ) + Text(event.method_name, style="yellow")
break
self.console.print(self.current_flow_tree) self.console.print(self.current_flow_tree)
self.console.print() self.console.print()
@crewai_event_bus.on(MethodExecutionFinishedEvent) @crewai_event_bus.on(MethodExecutionFinishedEvent)
def on_method_execution_finished(source, event: MethodExecutionFinishedEvent): def on_method_execution_finished(source, event: MethodExecutionFinishedEvent):
if self.current_method_branch: if self.verbose:
# Update method status if self.current_method_branch:
self.current_method_branch.label = Text( # Update method status
"✅ Completed: ", style="green bold" self.current_method_branch.label = Text(
) + Text(event.method_name, style="green") "✅ Completed: ", style="green bold"
self.console.print(self.current_flow_tree) ) + Text(event.method_name, style="green")
self.console.print() self.console.print(self.current_flow_tree)
self.console.print()
@crewai_event_bus.on(MethodExecutionFailedEvent) @crewai_event_bus.on(MethodExecutionFailedEvent)
def on_method_execution_failed(source, event: MethodExecutionFailedEvent): def on_method_execution_failed(source, event: MethodExecutionFailedEvent):
if self.current_method_branch: if self.verbose:
# Update method status to show failure if self.current_method_branch:
self.current_method_branch.label = Text( self.current_method_branch.label = Text(
"❌ Failed: ", style="red bold" "❌ Failed: ", style="red bold"
) + Text(event.method_name, style="red") ) + Text(event.method_name, style="red")
self.console.print(self.current_flow_tree) self.console.print(self.current_flow_tree)
self.console.print() self.console.print()
# ----------- TOOL USAGE EVENTS ----------- # ----------- TOOL USAGE EVENTS -----------
@crewai_event_bus.on(ToolUsageStartedEvent) @crewai_event_bus.on(ToolUsageStartedEvent)
def on_tool_usage_started(source, event: ToolUsageStartedEvent): def on_tool_usage_started(source, event: ToolUsageStartedEvent):
# Create tool usage content if self.verbose:
tool_content = Text() # Create tool usage content
tool_content.append("🔧 Using ", style="yellow bold") tool_content = Text()
tool_content.append(event.tool_name, style="yellow") tool_content.append("🔧 Using ", style="yellow bold")
tool_content.append(event.tool_name, style="yellow")
# Add to tree under the agent branch # Add to tree under the agent branch
if self.current_agent_branch: if self.current_agent_branch:
self.current_tool_branch = self.current_agent_branch.add(tool_content) self.current_tool_branch = self.current_agent_branch.add(
self.console.print(self.current_crew_tree) tool_content
self.console.print() )
self.console.print(self.current_crew_tree)
self.console.print()
@crewai_event_bus.on(ToolUsageFinishedEvent) @crewai_event_bus.on(ToolUsageFinishedEvent)
def on_tool_usage_finished(source, event: ToolUsageFinishedEvent): def on_tool_usage_finished(source, event: ToolUsageFinishedEvent):
# Create completion content if self.verbose:
completion_content = Text() completion_content = Text()
completion_content.append("🔧 Used ", style="green bold") completion_content.append("🔧 Used ", style="green bold")
completion_content.append(event.tool_name, style="green") completion_content.append(event.tool_name, style="green")
# Update under the agent branch # Update under the agent branch
if self.current_tool_branch: if self.current_tool_branch:
self.current_tool_branch.label = completion_content self.current_tool_branch.label = completion_content
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
self.console.print() self.console.print()
@crewai_event_bus.on(ToolUsageErrorEvent) @crewai_event_bus.on(ToolUsageErrorEvent)
def on_tool_usage_error(source, event: ToolUsageErrorEvent): def on_tool_usage_error(source, event: ToolUsageErrorEvent):
# Create tool error content if self.verbose:
error_content = Text() error_content = Text()
error_content.append("🔧 Tool Failed: ", style="red bold") error_content.append("🔧 Tool Failed: ", style="red bold")
error_content.append(event.tool_name, style="red") error_content.append(event.tool_name, style="red")
# Update under the agent branch # Update under the agent branch
if self.current_tool_branch: if self.current_tool_branch:
self.current_tool_branch.label = error_content self.current_tool_branch.label = error_content
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
self.console.print()
# Show error panel
panel = self._create_panel(
Text(
f"Tool usage failed: {event.tool_name}: {event.error}",
style="red",
),
"Tool Error",
"red",
)
self.console.print(panel)
self.console.print() self.console.print()
# Show error panel
panel = self._create_panel(
Text(
f"Tool usage failed: {event.tool_name}: {event.error}", style="red"
),
"Tool Error",
"red",
)
self.console.print(panel)
self.console.print()
# ----------- LLM EVENTS ----------- # ----------- LLM EVENTS -----------
@crewai_event_bus.on(LLMCallStartedEvent) @crewai_event_bus.on(LLMCallStartedEvent)
def on_llm_call_started(source, event: LLMCallStartedEvent): def on_llm_call_started(source, event: LLMCallStartedEvent):
# Create simple LLM call content if self.verbose:
llm_content = Text() llm_content = Text()
llm_content.append("🧠 Thinking...", style="blue bold") llm_content.append("🧠 Thinking...", style="blue bold")
# Add to tree under the agent branch # Add to tree under the agent branch
if self.current_agent_branch: if self.current_agent_branch:
self.current_tool_branch = self.current_agent_branch.add(llm_content) self.current_tool_branch = self.current_agent_branch.add(
self.console.print(self.current_crew_tree) llm_content
self.console.print() )
self.console.print(self.current_crew_tree)
self.console.print()
@crewai_event_bus.on(LLMCallCompletedEvent) @crewai_event_bus.on(LLMCallCompletedEvent)
def on_llm_call_completed(source, event: LLMCallCompletedEvent): def on_llm_call_completed(source, event: LLMCallCompletedEvent):
# Create simple completion content if self.verbose:
completion_content = Text() completion_content = Text()
completion_content.append("✨ Done", style="green bold") completion_content.append("✨ Done", style="green bold")
# Update under the agent branch # Update under the agent branch
if self.current_tool_branch: if self.current_tool_branch:
self.current_tool_branch.label = completion_content self.current_tool_branch.label = completion_content
self.console.print(self.current_crew_tree) self.console.print(self.current_crew_tree)
self.console.print() self.console.print()
@crewai_event_bus.on(LLMCallFailedEvent) @crewai_event_bus.on(LLMCallFailedEvent)
def on_llm_call_failed(source, event: LLMCallFailedEvent): def on_llm_call_failed(source, event: LLMCallFailedEvent):
# Create LLM error content if self.verbose:
error_content = Text() error_content = Text()
error_content.append("❌ LLM Call Failed\n", style="red bold") error_content.append("❌ LLM Call Failed\n", style="red bold")
error_content.append("Error: ", style="white") error_content.append("Error: ", style="white")
error_content.append(str(event.error), style="red") error_content.append(str(event.error), style="red")
# Update under the agent branch if it exists # Update under the agent branch if it exists
if self.current_tool_branch: if self.current_tool_branch:
self.current_tool_branch.label = Text("❌ LLM Failed", style="red bold") self.current_tool_branch.label = Text(
self.console.print(self.current_crew_tree) "❌ LLM Failed", style="red bold"
)
self.console.print(self.current_crew_tree)
self.console.print()
# Show error panel
panel = self._create_panel(error_content, "LLM Error", "red")
self.console.print(panel)
self.console.print() self.console.print()
# Show error panel
panel = self._create_panel(error_content, "LLM Error", "red")
self.console.print(panel)
self.console.print()
@crewai_event_bus.on(LLMStreamChunkEvent) @crewai_event_bus.on(LLMStreamChunkEvent)
def on_llm_stream_chunk(source, event: LLMStreamChunkEvent): def on_llm_stream_chunk(source, event: LLMStreamChunkEvent):
self.text_stream.write(event.chunk) self.text_stream.write(event.chunk)
@@ -595,67 +612,72 @@ class EventListener(BaseEventListener):
event.eval_llm or "", event.eval_llm or "",
) )
# Create test content for panel if self.verbose:
content = Text() content = Text()
content.append("🧪 Starting Crew Test\n\n", style="blue bold") content.append("🧪 Starting Crew Test\n\n", style="blue bold")
content.append("Crew: ", style="white") content.append("Crew: ", style="white")
content.append(f"{event.crew_name}\n", style="blue") content.append(f"{event.crew_name}\n", style="blue")
content.append("ID: ", style="white") content.append("ID: ", style="white")
content.append(str(source.id), style="blue") content.append(str(source.id), style="blue")
content.append("\nIterations: ", style="white") content.append("\nIterations: ", style="white")
content.append(str(event.n_iterations), style="yellow") content.append(str(event.n_iterations), style="yellow")
panel = self._create_panel(content, "Test Execution", "blue") panel = self._create_panel(content, "Test Execution", "blue")
self.console.print() self.console.print()
self.console.print(panel) self.console.print(panel)
self.console.print() self.console.print()
# Create and display the test tree # Create and display the test tree
test_label = Text()
test_label.append("🧪 Test: ", style="blue bold")
test_label.append(event.crew_name or "Crew", style="blue")
test_label.append("\n Status: ", style="white")
test_label.append("In Progress", style="yellow")
self.current_flow_tree = Tree(test_label)
self.current_flow_tree.add(Text("🔄 Running tests...", style="yellow"))
self.console.print(self.current_flow_tree)
self.console.print()
@crewai_event_bus.on(CrewTestCompletedEvent)
def on_crew_test_completed(source, event: CrewTestCompletedEvent):
if self.current_flow_tree:
# Update test tree label to show completion
test_label = Text() test_label = Text()
test_label.append("✅ Test: ", style="green bold") test_label.append("🧪 Test: ", style="blue bold")
test_label.append(event.crew_name or "Crew", style="green") test_label.append(event.crew_name or "Crew", style="blue")
test_label.append("\n Status: ", style="white") test_label.append("\n Status: ", style="white")
test_label.append("Completed", style="green bold") test_label.append("In Progress", style="yellow")
self.current_flow_tree.label = test_label
# Update the running tests node self.current_flow_tree = Tree(test_label)
for child in self.current_flow_tree.children: self.current_flow_tree.add(Text("🔄 Running tests...", style="yellow"))
if "Running tests" in str(child.label):
child.label = Text(
"✅ Tests completed successfully", style="green"
)
self.console.print(self.current_flow_tree) self.console.print(self.current_flow_tree)
self.console.print() self.console.print()
# Create completion panel @crewai_event_bus.on(CrewTestCompletedEvent)
completion_content = Text() def on_crew_test_completed(source, event: CrewTestCompletedEvent):
completion_content.append("Test Execution Completed\n", style="green bold") if self.verbose:
completion_content.append("Crew: ", style="white") if self.current_flow_tree:
completion_content.append(f"{event.crew_name}\n", style="green") # Update test tree label to show completion
completion_content.append("Status: ", style="white") test_label = Text()
completion_content.append("All tests passed", style="green") test_label.append("✅ Test: ", style="green bold")
test_label.append(event.crew_name or "Crew", style="green")
test_label.append("\n Status: ", style="white")
test_label.append("Completed", style="green bold")
self.current_flow_tree.label = test_label
panel = self._create_panel(completion_content, "Test Completion", "green") # Update the running tests node
self.console.print(panel) for child in self.current_flow_tree.children:
self.console.print() if "Running tests" in str(child.label):
child.label = Text(
"✅ Tests completed successfully", style="green"
)
self.console.print(self.current_flow_tree)
self.console.print()
# Create completion panel
completion_content = Text()
completion_content.append(
"Test Execution Completed\n", style="green bold"
)
completion_content.append("Crew: ", style="white")
completion_content.append(f"{event.crew_name}\n", style="green")
completion_content.append("Status: ", style="white")
completion_content.append("All tests passed", style="green")
panel = self._create_panel(
completion_content, "Test Completion", "green"
)
self.console.print(panel)
self.console.print()
event_listener = EventListener() event_listener = EventListener()