mirror of
https://github.com/crewAIInc/crewAI.git
synced 2025-12-16 04:18:35 +00:00
Compare commits
3 Commits
1.6.1
...
devin/1750
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
13cd8baa2c | ||
|
|
a9823a4368 | ||
|
|
4e288e4cf7 |
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from rich.console import Console
|
||||
@@ -7,6 +8,39 @@ from rich.tree import Tree
|
||||
from rich.live import Live
|
||||
from rich.syntax import Syntax
|
||||
|
||||
DEFAULT_EMOJI_MAP = {
|
||||
"✅": "[DONE]",
|
||||
"❌": "[FAILED]",
|
||||
"🚀": "[CREW]",
|
||||
"🔄": "[RUNNING]",
|
||||
"📋": "[TASK]",
|
||||
"🔧": "[TOOL]",
|
||||
"🧠": "[THINKING]",
|
||||
"🌊": "[FLOW]",
|
||||
"✨": "[CREATED]",
|
||||
"🧪": "[TEST]",
|
||||
"📚": "[KNOWLEDGE]",
|
||||
"🔍": "[SEARCH]",
|
||||
"🔎": "[QUERY]",
|
||||
"🤖": "[AGENT]",
|
||||
"📊": "[METRICS]",
|
||||
"⚡": "[QUICK]",
|
||||
"🎯": "[TARGET]",
|
||||
"🔗": "[LINK]",
|
||||
"💡": "[IDEA]",
|
||||
"⚠️": "[WARNING]",
|
||||
"🎉": "[SUCCESS]",
|
||||
"🔥": "[HOT]",
|
||||
"💾": "[SAVE]",
|
||||
"🔒": "[SECURE]",
|
||||
"🌟": "[STAR]",
|
||||
}
|
||||
|
||||
|
||||
def _parse_bool_env(val: str) -> bool:
|
||||
"""Parse environment variable value to boolean."""
|
||||
return val.lower() in ("true", "1", "yes") if val else False
|
||||
|
||||
|
||||
class ConsoleFormatter:
|
||||
current_crew_tree: Optional[Tree] = None
|
||||
@@ -24,6 +58,13 @@ class ConsoleFormatter:
|
||||
def __init__(self, verbose: bool = False):
|
||||
self.console = Console(width=None)
|
||||
self.verbose = verbose
|
||||
self.disable_emojis = _parse_bool_env(os.getenv("CREWAI_DISABLE_EMOJIS", ""))
|
||||
|
||||
self.emoji_map = DEFAULT_EMOJI_MAP.copy()
|
||||
self.icon_cache: Dict[str, str] = {}
|
||||
if self.disable_emojis:
|
||||
self.icon_cache = {emoji: text for emoji, text in self.emoji_map.items()}
|
||||
|
||||
# Live instance to dynamically update a Tree renderable (e.g. the Crew tree)
|
||||
# When multiple Tree objects are printed sequentially we reuse this Live
|
||||
# instance so the previous render is replaced instead of writing a new one.
|
||||
@@ -31,6 +72,17 @@ class ConsoleFormatter:
|
||||
# final Tree persists on the terminal.
|
||||
self._live: Optional[Live] = None
|
||||
|
||||
def _get_icon(self, emoji: str) -> str:
|
||||
"""Get emoji or text alternative based on disable_emojis setting."""
|
||||
if self.disable_emojis:
|
||||
if emoji in self.icon_cache:
|
||||
return self.icon_cache[emoji]
|
||||
ascii_fallback = emoji.encode('ascii', 'ignore').decode('ascii')
|
||||
fallback = f"[ICON:{ascii_fallback or 'UNKNOWN'}]"
|
||||
self.icon_cache[emoji] = fallback
|
||||
return fallback
|
||||
return emoji
|
||||
|
||||
def create_panel(self, content: Text, title: str, style: str = "blue") -> Panel:
|
||||
"""Create a standardized panel with consistent styling."""
|
||||
return Panel(
|
||||
@@ -167,15 +219,15 @@ class ConsoleFormatter:
|
||||
return
|
||||
|
||||
if status == "completed":
|
||||
prefix, style = "✅ Crew:", "green"
|
||||
prefix, style = f"{self._get_icon('✅')} Crew:", "green"
|
||||
title = "Crew Completion"
|
||||
content_title = "Crew Execution Completed"
|
||||
elif status == "failed":
|
||||
prefix, style = "❌ Crew:", "red"
|
||||
prefix, style = f"{self._get_icon('❌')} Crew:", "red"
|
||||
title = "Crew Failure"
|
||||
content_title = "Crew Execution Failed"
|
||||
else:
|
||||
prefix, style = "🚀 Crew:", "cyan"
|
||||
prefix, style = f"{self._get_icon('🚀')} Crew:", "cyan"
|
||||
title = "Crew Execution"
|
||||
content_title = "Crew Execution Started"
|
||||
|
||||
@@ -202,7 +254,7 @@ class ConsoleFormatter:
|
||||
return None
|
||||
|
||||
tree = Tree(
|
||||
Text("🚀 Crew: ", style="cyan bold") + Text(crew_name, style="cyan")
|
||||
Text(f"{self._get_icon('🚀')} Crew: ", style="cyan bold") + Text(crew_name, style="cyan")
|
||||
)
|
||||
|
||||
content = self.create_status_content(
|
||||
@@ -227,7 +279,7 @@ class ConsoleFormatter:
|
||||
return None
|
||||
|
||||
task_content = Text()
|
||||
task_content.append(f"📋 Task: {task_id}", style="yellow bold")
|
||||
task_content.append(f"{self._get_icon('📋')} Task: {task_id}", style="yellow bold")
|
||||
task_content.append("\nStatus: ", style="white")
|
||||
task_content.append("Executing Task...", style="yellow dim")
|
||||
|
||||
@@ -258,11 +310,11 @@ class ConsoleFormatter:
|
||||
|
||||
if status == "completed":
|
||||
style = "green"
|
||||
status_text = "✅ Completed"
|
||||
status_text = f"{self._get_icon('✅')} Completed"
|
||||
panel_title = "Task Completion"
|
||||
else:
|
||||
style = "red"
|
||||
status_text = "❌ Failed"
|
||||
status_text = f"{self._get_icon('❌')} Failed"
|
||||
panel_title = "Task Failure"
|
||||
|
||||
# Update tree label
|
||||
@@ -271,7 +323,7 @@ class ConsoleFormatter:
|
||||
# Build label without introducing stray blank lines
|
||||
task_content = Text()
|
||||
# First line: Task ID
|
||||
task_content.append(f"📋 Task: {task_id}", style=f"{style} bold")
|
||||
task_content.append(f"{self._get_icon('📋')} Task: {task_id}", style=f"{style} bold")
|
||||
|
||||
# Second line: Assigned to
|
||||
task_content.append("\nAssigned to: ", style="white")
|
||||
@@ -330,14 +382,14 @@ class ConsoleFormatter:
|
||||
|
||||
# Create initial tree with flow ID
|
||||
flow_label = Text()
|
||||
flow_label.append("🌊 Flow: ", style="blue bold")
|
||||
flow_label.append(f"{self._get_icon('🌊')} Flow: ", style="blue bold")
|
||||
flow_label.append(flow_name, style="blue")
|
||||
flow_label.append("\nID: ", style="white")
|
||||
flow_label.append(flow_id, style="blue")
|
||||
|
||||
flow_tree = Tree(flow_label)
|
||||
self.add_tree_node(flow_tree, "✨ Created", "blue")
|
||||
self.add_tree_node(flow_tree, "✅ Initialization Complete", "green")
|
||||
self.add_tree_node(flow_tree, f"{self._get_icon('✨')} Created", "blue")
|
||||
self.add_tree_node(flow_tree, f"{self._get_icon('✅')} Initialization Complete", "green")
|
||||
|
||||
return flow_tree
|
||||
|
||||
@@ -345,13 +397,13 @@ class ConsoleFormatter:
|
||||
"""Initialize a flow execution tree."""
|
||||
flow_tree = Tree("")
|
||||
flow_label = Text()
|
||||
flow_label.append("🌊 Flow: ", style="blue bold")
|
||||
flow_label.append(f"{self._get_icon('🌊')} Flow: ", style="blue bold")
|
||||
flow_label.append(flow_name, style="blue")
|
||||
flow_label.append("\nID: ", style="white")
|
||||
flow_label.append(flow_id, style="blue")
|
||||
flow_tree.label = flow_label
|
||||
|
||||
self.add_tree_node(flow_tree, "🧠 Starting Flow...", "yellow")
|
||||
self.add_tree_node(flow_tree, f"{self._get_icon('🧠')} Starting Flow...", "yellow")
|
||||
|
||||
self.print(flow_tree)
|
||||
self.print()
|
||||
@@ -373,7 +425,7 @@ class ConsoleFormatter:
|
||||
# Update main flow label
|
||||
self.update_tree_label(
|
||||
flow_tree,
|
||||
"✅ Flow Finished:" if status == "completed" else "❌ Flow Failed:",
|
||||
f"{self._get_icon('✅')} Flow Finished:" if status == "completed" else f"{self._get_icon('❌')} Flow Failed:",
|
||||
flow_name,
|
||||
"green" if status == "completed" else "red",
|
||||
)
|
||||
@@ -383,9 +435,9 @@ class ConsoleFormatter:
|
||||
if "Starting Flow" in str(child.label):
|
||||
child.label = Text(
|
||||
(
|
||||
"✅ Flow Completed"
|
||||
f"{self._get_icon('✅')} Flow Completed"
|
||||
if status == "completed"
|
||||
else "❌ Flow Failed"
|
||||
else f"{self._get_icon('❌')} Flow Failed"
|
||||
),
|
||||
style="green" if status == "completed" else "red",
|
||||
)
|
||||
@@ -418,20 +470,20 @@ class ConsoleFormatter:
|
||||
return None
|
||||
|
||||
if status == "running":
|
||||
prefix, style = "🔄 Running:", "yellow"
|
||||
prefix, style = f"{self._get_icon('🔄')} Running:", "yellow"
|
||||
elif status == "completed":
|
||||
prefix, style = "✅ Completed:", "green"
|
||||
prefix, style = f"{self._get_icon('✅')} Completed:", "green"
|
||||
# Update initialization node when a method completes successfully
|
||||
for child in flow_tree.children:
|
||||
if "Starting Flow" in str(child.label):
|
||||
child.label = Text("Flow Method Step", style="white")
|
||||
break
|
||||
else:
|
||||
prefix, style = "❌ Failed:", "red"
|
||||
prefix, style = f"{self._get_icon('❌')} Failed:", "red"
|
||||
# Update initialization node on failure
|
||||
for child in flow_tree.children:
|
||||
if "Starting Flow" in str(child.label):
|
||||
child.label = Text("❌ Flow Step Failed", style="red")
|
||||
child.label = Text(f"{self._get_icon('❌')} Flow Step Failed", style="red")
|
||||
break
|
||||
|
||||
if not method_branch:
|
||||
@@ -453,7 +505,7 @@ class ConsoleFormatter:
|
||||
|
||||
def get_llm_tree(self, tool_name: str):
|
||||
text = Text()
|
||||
text.append(f"🔧 Using {tool_name} from LLM available_function", style="yellow")
|
||||
text.append(f"{self._get_icon('🔧')} Using {tool_name} from LLM available_function", style="yellow")
|
||||
|
||||
tree = self.current_flow_tree or self.current_crew_tree
|
||||
|
||||
@@ -484,7 +536,7 @@ class ConsoleFormatter:
|
||||
tool_name: str,
|
||||
):
|
||||
tree = self.get_llm_tree(tool_name)
|
||||
self.add_tree_node(tree, "✅ Tool Usage Completed", "green")
|
||||
self.add_tree_node(tree, f"{self._get_icon('✅')} Tool Usage Completed", "green")
|
||||
self.print(tree)
|
||||
self.print()
|
||||
|
||||
@@ -494,7 +546,7 @@ class ConsoleFormatter:
|
||||
error: str,
|
||||
):
|
||||
tree = self.get_llm_tree(tool_name)
|
||||
self.add_tree_node(tree, "❌ Tool Usage Failed", "red")
|
||||
self.add_tree_node(tree, f"{self._get_icon('❌')} Tool Usage Failed", "red")
|
||||
self.print(tree)
|
||||
self.print()
|
||||
|
||||
@@ -541,7 +593,7 @@ class ConsoleFormatter:
|
||||
# Update label with current count
|
||||
self.update_tree_label(
|
||||
tool_branch,
|
||||
"🔧",
|
||||
self._get_icon("🔧"),
|
||||
f"Using {tool_name} ({self.tool_usage_counts[tool_name]})",
|
||||
"yellow",
|
||||
)
|
||||
@@ -570,7 +622,7 @@ class ConsoleFormatter:
|
||||
# Update the existing tool node's label
|
||||
self.update_tree_label(
|
||||
tool_branch,
|
||||
"🔧",
|
||||
self._get_icon("🔧"),
|
||||
f"Used {tool_name} ({self.tool_usage_counts[tool_name]})",
|
||||
"green",
|
||||
)
|
||||
@@ -600,7 +652,7 @@ class ConsoleFormatter:
|
||||
if tool_branch:
|
||||
self.update_tree_label(
|
||||
tool_branch,
|
||||
"🔧 Failed",
|
||||
f"{self._get_icon('🔧')} Failed",
|
||||
f"{tool_name} ({self.tool_usage_counts[tool_name]})",
|
||||
"red",
|
||||
)
|
||||
@@ -646,7 +698,7 @@ class ConsoleFormatter:
|
||||
|
||||
if should_add_thinking:
|
||||
tool_branch = branch_to_use.add("")
|
||||
self.update_tree_label(tool_branch, "🧠", "Thinking...", "blue")
|
||||
self.update_tree_label(tool_branch, self._get_icon("🧠"), "Thinking...", "blue")
|
||||
self.current_tool_branch = tool_branch
|
||||
self.print(tree_to_use)
|
||||
self.print()
|
||||
@@ -756,7 +808,7 @@ class ConsoleFormatter:
|
||||
|
||||
# Update the thinking branch to show failure
|
||||
if thinking_branch_to_update:
|
||||
thinking_branch_to_update.label = Text("❌ LLM Failed", style="red bold")
|
||||
thinking_branch_to_update.label = Text(f"{self._get_icon('❌')} LLM Failed", style="red bold")
|
||||
# Clear the current_tool_branch reference
|
||||
if self.current_tool_branch is thinking_branch_to_update:
|
||||
self.current_tool_branch = None
|
||||
@@ -766,7 +818,7 @@ class ConsoleFormatter:
|
||||
|
||||
# Show error panel
|
||||
error_content = Text()
|
||||
error_content.append("❌ LLM Call Failed\n", style="red bold")
|
||||
error_content.append(f"{self._get_icon('❌')} LLM Call Failed\n", style="red bold")
|
||||
error_content.append("Error: ", style="white")
|
||||
error_content.append(str(error), style="red")
|
||||
|
||||
@@ -781,7 +833,7 @@ class ConsoleFormatter:
|
||||
|
||||
# Create initial panel
|
||||
content = Text()
|
||||
content.append("🧪 Starting Crew Test\n\n", style="blue bold")
|
||||
content.append(f"{self._get_icon('🧪')} Starting Crew Test\n\n", style="blue bold")
|
||||
content.append("Crew: ", style="white")
|
||||
content.append(f"{crew_name}\n", style="blue")
|
||||
content.append("ID: ", style="white")
|
||||
@@ -795,13 +847,13 @@ class ConsoleFormatter:
|
||||
|
||||
# Create and display the test tree
|
||||
test_label = Text()
|
||||
test_label.append("🧪 Test: ", style="blue bold")
|
||||
test_label.append(f"{self._get_icon('🧪')} Test: ", style="blue bold")
|
||||
test_label.append(crew_name or "Crew", style="blue")
|
||||
test_label.append("\nStatus: ", style="white")
|
||||
test_label.append("In Progress", style="yellow")
|
||||
|
||||
test_tree = Tree(test_label)
|
||||
self.add_tree_node(test_tree, "🔄 Running tests...", "yellow")
|
||||
self.add_tree_node(test_tree, f"{self._get_icon('🔄')} Running tests...", "yellow")
|
||||
|
||||
self.print(test_tree)
|
||||
self.print()
|
||||
@@ -817,7 +869,7 @@ class ConsoleFormatter:
|
||||
if flow_tree:
|
||||
# Update test tree label to show completion
|
||||
test_label = Text()
|
||||
test_label.append("✅ Test: ", style="green bold")
|
||||
test_label.append(f"{self._get_icon('✅')} Test: ", style="green bold")
|
||||
test_label.append(crew_name or "Crew", style="green")
|
||||
test_label.append("\nStatus: ", style="white")
|
||||
test_label.append("Completed", style="green bold")
|
||||
@@ -826,7 +878,7 @@ class ConsoleFormatter:
|
||||
# Update the running tests node
|
||||
for child in flow_tree.children:
|
||||
if "Running tests" in str(child.label):
|
||||
child.label = Text("✅ Tests completed successfully", style="green")
|
||||
child.label = Text(f"{self._get_icon('✅')} Tests completed successfully", style="green")
|
||||
break
|
||||
|
||||
self.print(flow_tree)
|
||||
@@ -848,7 +900,7 @@ class ConsoleFormatter:
|
||||
return
|
||||
|
||||
content = Text()
|
||||
content.append("📋 Crew Training Started\n", style="blue bold")
|
||||
content.append(f"{self._get_icon('📋')} Crew Training Started\n", style="blue bold")
|
||||
content.append("Crew: ", style="white")
|
||||
content.append(f"{crew_name}\n", style="blue")
|
||||
content.append("Time: ", style="white")
|
||||
@@ -863,7 +915,7 @@ class ConsoleFormatter:
|
||||
return
|
||||
|
||||
content = Text()
|
||||
content.append("✅ Crew Training Completed\n", style="green bold")
|
||||
content.append(f"{self._get_icon('✅')} Crew Training Completed\n", style="green bold")
|
||||
content.append("Crew: ", style="white")
|
||||
content.append(f"{crew_name}\n", style="green")
|
||||
content.append("Time: ", style="white")
|
||||
@@ -878,7 +930,7 @@ class ConsoleFormatter:
|
||||
return
|
||||
|
||||
failure_content = Text()
|
||||
failure_content.append("❌ Crew Training Failed\n", style="red bold")
|
||||
failure_content.append(f"{self._get_icon('❌')} Crew Training Failed\n", style="red bold")
|
||||
failure_content.append("Crew: ", style="white")
|
||||
failure_content.append(crew_name or "Crew", style="red")
|
||||
|
||||
@@ -891,7 +943,7 @@ class ConsoleFormatter:
|
||||
return
|
||||
|
||||
failure_content = Text()
|
||||
failure_content.append("❌ Crew Test Failed\n", style="red bold")
|
||||
failure_content.append(f"{self._get_icon('❌')} Crew Test Failed\n", style="red bold")
|
||||
failure_content.append("Crew: ", style="white")
|
||||
failure_content.append(crew_name or "Crew", style="red")
|
||||
|
||||
@@ -906,7 +958,7 @@ class ConsoleFormatter:
|
||||
# Create initial tree for LiteAgent if it doesn't exist
|
||||
if not self.current_lite_agent_branch:
|
||||
lite_agent_label = Text()
|
||||
lite_agent_label.append("🤖 LiteAgent: ", style="cyan bold")
|
||||
lite_agent_label.append(f"{self._get_icon('🤖')} LiteAgent: ", style="cyan bold")
|
||||
lite_agent_label.append(lite_agent_role, style="cyan")
|
||||
lite_agent_label.append("\nStatus: ", style="white")
|
||||
lite_agent_label.append("In Progress", style="yellow")
|
||||
@@ -931,15 +983,15 @@ class ConsoleFormatter:
|
||||
|
||||
# Determine style based on status
|
||||
if status == "completed":
|
||||
prefix, style = "✅ LiteAgent:", "green"
|
||||
prefix, style = f"{self._get_icon('✅')} LiteAgent:", "green"
|
||||
status_text = "Completed"
|
||||
title = "LiteAgent Completion"
|
||||
elif status == "failed":
|
||||
prefix, style = "❌ LiteAgent:", "red"
|
||||
prefix, style = f"{self._get_icon('❌')} LiteAgent:", "red"
|
||||
status_text = "Failed"
|
||||
title = "LiteAgent Error"
|
||||
else:
|
||||
prefix, style = "🤖 LiteAgent:", "yellow"
|
||||
prefix, style = f"{self._get_icon('🤖')} LiteAgent:", "yellow"
|
||||
status_text = "In Progress"
|
||||
title = "LiteAgent Status"
|
||||
|
||||
@@ -1010,7 +1062,7 @@ class ConsoleFormatter:
|
||||
|
||||
knowledge_branch = branch_to_use.add("")
|
||||
self.update_tree_label(
|
||||
knowledge_branch, "🔍", "Knowledge Retrieval Started", "blue"
|
||||
knowledge_branch, self._get_icon("🔍"), "Knowledge Retrieval Started", "blue"
|
||||
)
|
||||
|
||||
self.print(tree_to_use)
|
||||
@@ -1041,7 +1093,7 @@ class ConsoleFormatter:
|
||||
|
||||
knowledge_panel = Panel(
|
||||
Text(knowledge_text, style="white"),
|
||||
title="📚 Retrieved Knowledge",
|
||||
title=f"{self._get_icon('📚')} Retrieved Knowledge",
|
||||
border_style="green",
|
||||
padding=(1, 2),
|
||||
)
|
||||
@@ -1053,7 +1105,7 @@ class ConsoleFormatter:
|
||||
for child in branch_to_use.children:
|
||||
if "Knowledge Retrieval Started" in str(child.label):
|
||||
self.update_tree_label(
|
||||
child, "✅", "Knowledge Retrieval Completed", "green"
|
||||
child, self._get_icon("✅"), "Knowledge Retrieval Completed", "green"
|
||||
)
|
||||
knowledge_branch_found = True
|
||||
break
|
||||
@@ -1066,7 +1118,7 @@ class ConsoleFormatter:
|
||||
and "Completed" not in str(child.label)
|
||||
):
|
||||
self.update_tree_label(
|
||||
child, "✅", "Knowledge Retrieval Completed", "green"
|
||||
child, self._get_icon("✅"), "Knowledge Retrieval Completed", "green"
|
||||
)
|
||||
knowledge_branch_found = True
|
||||
break
|
||||
@@ -1074,7 +1126,7 @@ class ConsoleFormatter:
|
||||
if not knowledge_branch_found:
|
||||
knowledge_branch = branch_to_use.add("")
|
||||
self.update_tree_label(
|
||||
knowledge_branch, "✅", "Knowledge Retrieval Completed", "green"
|
||||
knowledge_branch, self._get_icon("✅"), "Knowledge Retrieval Completed", "green"
|
||||
)
|
||||
|
||||
self.print(tree_to_use)
|
||||
@@ -1086,7 +1138,7 @@ class ConsoleFormatter:
|
||||
|
||||
knowledge_panel = Panel(
|
||||
Text(knowledge_text, style="white"),
|
||||
title="📚 Retrieved Knowledge",
|
||||
title=f"{self._get_icon('📚')} Retrieved Knowledge",
|
||||
border_style="green",
|
||||
padding=(1, 2),
|
||||
)
|
||||
@@ -1111,7 +1163,7 @@ class ConsoleFormatter:
|
||||
|
||||
query_branch = branch_to_use.add("")
|
||||
self.update_tree_label(
|
||||
query_branch, "🔎", f"Query: {task_prompt[:50]}...", "yellow"
|
||||
query_branch, self._get_icon("🔎"), f"Query: {task_prompt[:50]}...", "yellow"
|
||||
)
|
||||
|
||||
self.print(tree_to_use)
|
||||
@@ -1132,7 +1184,7 @@ class ConsoleFormatter:
|
||||
|
||||
if branch_to_use and tree_to_use:
|
||||
query_branch = branch_to_use.add("")
|
||||
self.update_tree_label(query_branch, "❌", "Knowledge Query Failed", "red")
|
||||
self.update_tree_label(query_branch, self._get_icon("❌"), "Knowledge Query Failed", "red")
|
||||
self.print(tree_to_use)
|
||||
self.print()
|
||||
|
||||
@@ -1158,7 +1210,7 @@ class ConsoleFormatter:
|
||||
return None
|
||||
|
||||
query_branch = branch_to_use.add("")
|
||||
self.update_tree_label(query_branch, "✅", "Knowledge Query Completed", "green")
|
||||
self.update_tree_label(query_branch, self._get_icon("✅"), "Knowledge Query Completed", "green")
|
||||
|
||||
self.print(tree_to_use)
|
||||
self.print()
|
||||
@@ -1178,7 +1230,7 @@ class ConsoleFormatter:
|
||||
|
||||
if branch_to_use and tree_to_use:
|
||||
query_branch = branch_to_use.add("")
|
||||
self.update_tree_label(query_branch, "❌", "Knowledge Search Failed", "red")
|
||||
self.update_tree_label(query_branch, self._get_icon("❌"), "Knowledge Search Failed", "red")
|
||||
self.print(tree_to_use)
|
||||
self.print()
|
||||
|
||||
@@ -1223,7 +1275,7 @@ class ConsoleFormatter:
|
||||
status_text = (
|
||||
f"Reasoning (Attempt {attempt})" if attempt > 1 else "Reasoning..."
|
||||
)
|
||||
self.update_tree_label(reasoning_branch, "🧠", status_text, "blue")
|
||||
self.update_tree_label(reasoning_branch, self._get_icon("🧠"), status_text, "blue")
|
||||
|
||||
self.print(tree_to_use)
|
||||
self.print()
|
||||
@@ -1254,7 +1306,7 @@ class ConsoleFormatter:
|
||||
)
|
||||
|
||||
if reasoning_branch is not None:
|
||||
self.update_tree_label(reasoning_branch, "✅", status_text, style)
|
||||
self.update_tree_label(reasoning_branch, self._get_icon("✅"), status_text, style)
|
||||
|
||||
if tree_to_use is not None:
|
||||
self.print(tree_to_use)
|
||||
@@ -1263,7 +1315,7 @@ class ConsoleFormatter:
|
||||
if plan:
|
||||
plan_panel = Panel(
|
||||
Text(plan, style="white"),
|
||||
title="🧠 Reasoning Plan",
|
||||
title=f"{self._get_icon('🧠')} Reasoning Plan",
|
||||
border_style=style,
|
||||
padding=(1, 2),
|
||||
)
|
||||
@@ -1292,7 +1344,7 @@ class ConsoleFormatter:
|
||||
)
|
||||
|
||||
if reasoning_branch is not None:
|
||||
self.update_tree_label(reasoning_branch, "❌", "Reasoning Failed", "red")
|
||||
self.update_tree_label(reasoning_branch, self._get_icon("❌"), "Reasoning Failed", "red")
|
||||
|
||||
if tree_to_use is not None:
|
||||
self.print(tree_to_use)
|
||||
@@ -1335,7 +1387,7 @@ class ConsoleFormatter:
|
||||
# Create and display the panel
|
||||
agent_panel = Panel(
|
||||
content,
|
||||
title="🤖 Agent Started",
|
||||
title=f"{self._get_icon('🤖')} Agent Started",
|
||||
border_style="magenta",
|
||||
padding=(1, 2),
|
||||
)
|
||||
@@ -1406,7 +1458,7 @@ class ConsoleFormatter:
|
||||
# Create the main action panel
|
||||
action_panel = Panel(
|
||||
main_content,
|
||||
title="🔧 Agent Tool Execution",
|
||||
title=f"{self._get_icon('🔧')} Agent Tool Execution",
|
||||
border_style="magenta",
|
||||
padding=(1, 2),
|
||||
)
|
||||
@@ -1448,7 +1500,7 @@ class ConsoleFormatter:
|
||||
# Create and display the finish panel
|
||||
finish_panel = Panel(
|
||||
content,
|
||||
title="✅ Agent Final Answer",
|
||||
title=f"{self._get_icon('✅')} Agent Final Answer",
|
||||
border_style="green",
|
||||
padding=(1, 2),
|
||||
)
|
||||
|
||||
1
tests/utilities/events/utils/__init__.py
Normal file
1
tests/utilities/events/utils/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
from rich.tree import Tree
|
||||
|
||||
from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
|
||||
|
||||
|
||||
class TestConsoleFormatterEmojiDisable:
|
||||
"""Test emoji disable functionality in ConsoleFormatter."""
|
||||
|
||||
def test_emoji_enabled_by_default(self):
|
||||
"""Test that emojis are enabled by default."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
assert formatter._get_icon("❌") == "❌"
|
||||
assert formatter._get_icon("🚀") == "🚀"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_emoji_disabled_with_true(self):
|
||||
"""Test that emojis are disabled when CREWAI_DISABLE_EMOJIS=true."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "[DONE]"
|
||||
assert formatter._get_icon("❌") == "[FAILED]"
|
||||
assert formatter._get_icon("🚀") == "[CREW]"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "1"})
|
||||
def test_emoji_disabled_with_one(self):
|
||||
"""Test that emojis are disabled when CREWAI_DISABLE_EMOJIS=1."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "[DONE]"
|
||||
assert formatter._get_icon("❌") == "[FAILED]"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "yes"})
|
||||
def test_emoji_disabled_with_yes(self):
|
||||
"""Test that emojis are disabled when CREWAI_DISABLE_EMOJIS=yes."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "[DONE]"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "false"})
|
||||
def test_emoji_enabled_with_false(self):
|
||||
"""Test that emojis are enabled when CREWAI_DISABLE_EMOJIS=false."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "0"})
|
||||
def test_emoji_enabled_with_zero(self):
|
||||
"""Test that emojis are enabled when CREWAI_DISABLE_EMOJIS=0."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": ""})
|
||||
def test_emoji_enabled_with_empty_string(self):
|
||||
"""Test that emojis are enabled when CREWAI_DISABLE_EMOJIS is empty."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
|
||||
def test_emoji_enabled_when_env_var_not_set(self):
|
||||
"""Test that emojis are enabled when CREWAI_DISABLE_EMOJIS is not set."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
|
||||
def test_all_emoji_mappings(self):
|
||||
"""Test that all emojis in EMOJI_MAP have proper text alternatives."""
|
||||
with patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"}):
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
|
||||
expected_mappings = {
|
||||
"✅": "[DONE]",
|
||||
"❌": "[FAILED]",
|
||||
"🚀": "[CREW]",
|
||||
"🔄": "[RUNNING]",
|
||||
"📋": "[TASK]",
|
||||
"🔧": "[TOOL]",
|
||||
"🧠": "[THINKING]",
|
||||
"🌊": "[FLOW]",
|
||||
"✨": "[CREATED]",
|
||||
"🧪": "[TEST]",
|
||||
"📚": "[KNOWLEDGE]",
|
||||
"🔍": "[SEARCH]",
|
||||
"🔎": "[QUERY]",
|
||||
"🤖": "[AGENT]",
|
||||
}
|
||||
|
||||
for emoji, expected_text in expected_mappings.items():
|
||||
assert formatter._get_icon(emoji) == expected_text
|
||||
|
||||
def test_unknown_emoji_fallback(self):
|
||||
"""Test that unknown emojis fall back to proper representation."""
|
||||
with patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"}):
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
result = formatter._get_icon("🦄")
|
||||
assert result == "[ICON:UNKNOWN]"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_crew_tree_creation_without_emojis(self):
|
||||
"""Test that crew tree creation works without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
tree = formatter.create_crew_tree("Test Crew", "test-id")
|
||||
|
||||
assert tree is not None
|
||||
tree_label_str = str(tree.label)
|
||||
assert "[CREW]" in tree_label_str
|
||||
assert "🚀" not in tree_label_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_task_branch_creation_without_emojis(self):
|
||||
"""Test that task branch creation works without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
crew_tree = Tree("Test Crew")
|
||||
task_branch = formatter.create_task_branch(crew_tree, "test-task-id")
|
||||
|
||||
assert task_branch is not None
|
||||
task_label_str = str(task_branch.label)
|
||||
assert "[TASK]" in task_label_str
|
||||
assert "📋" not in task_label_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_update_crew_tree_completed_without_emojis(self):
|
||||
"""Test that crew tree completion updates work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
tree = Tree("Test")
|
||||
|
||||
formatter.update_crew_tree(tree, "Test Crew", "test-id", "completed", "Test output")
|
||||
|
||||
tree_str = str(tree.label)
|
||||
assert "[DONE]" in tree_str
|
||||
assert "✅" not in tree_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_tool_usage_without_emojis(self):
|
||||
"""Test that tool usage events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
crew_tree = Tree("Test Crew")
|
||||
agent_branch = crew_tree.add("Test Agent")
|
||||
|
||||
formatter.handle_tool_usage_started(agent_branch, "test_tool", crew_tree)
|
||||
|
||||
found_tool_branch = False
|
||||
for child in agent_branch.children:
|
||||
child_str = str(child.label)
|
||||
if "test_tool" in child_str:
|
||||
assert "[TOOL]" in child_str
|
||||
assert "🔧" not in child_str
|
||||
found_tool_branch = True
|
||||
break
|
||||
|
||||
assert found_tool_branch, "Tool branch should have been created"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_llm_call_without_emojis(self):
|
||||
"""Test that LLM call events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
crew_tree = Tree("Test Crew")
|
||||
agent_branch = crew_tree.add("Test Agent")
|
||||
|
||||
thinking_branch = formatter.handle_llm_call_started(agent_branch, crew_tree)
|
||||
|
||||
if thinking_branch:
|
||||
thinking_str = str(thinking_branch.label)
|
||||
assert "[THINKING]" in thinking_str
|
||||
assert "🧠" not in thinking_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_knowledge_retrieval_without_emojis(self):
|
||||
"""Test that knowledge retrieval events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
crew_tree = Tree("Test Crew")
|
||||
agent_branch = crew_tree.add("Test Agent")
|
||||
|
||||
knowledge_branch = formatter.handle_knowledge_retrieval_started(agent_branch, crew_tree)
|
||||
|
||||
if knowledge_branch:
|
||||
knowledge_str = str(knowledge_branch.label)
|
||||
assert "[SEARCH]" in knowledge_str
|
||||
assert "🔍" not in knowledge_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_reasoning_without_emojis(self):
|
||||
"""Test that reasoning events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
crew_tree = Tree("Test Crew")
|
||||
agent_branch = crew_tree.add("Test Agent")
|
||||
|
||||
reasoning_branch = formatter.handle_reasoning_started(agent_branch, 1, crew_tree)
|
||||
|
||||
if reasoning_branch:
|
||||
reasoning_str = str(reasoning_branch.label)
|
||||
assert "[THINKING]" in reasoning_str
|
||||
assert "🧠" not in reasoning_str
|
||||
|
||||
def test_case_insensitive_environment_variable(self):
|
||||
"""Test that environment variable parsing is case insensitive."""
|
||||
test_cases = [
|
||||
("TRUE", True),
|
||||
("True", True),
|
||||
("true", True),
|
||||
("1", True),
|
||||
("YES", True),
|
||||
("yes", True),
|
||||
("Yes", True),
|
||||
("FALSE", False),
|
||||
("False", False),
|
||||
("false", False),
|
||||
("0", False),
|
||||
("NO", False),
|
||||
("no", False),
|
||||
("No", False),
|
||||
("", False),
|
||||
("random", False),
|
||||
]
|
||||
|
||||
for env_value, expected_disabled in test_cases:
|
||||
with patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": env_value}):
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
assert formatter.disable_emojis == expected_disabled, f"Failed for env_value: {env_value}"
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_flow_events_without_emojis(self):
|
||||
"""Test that flow events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
|
||||
flow_tree = formatter.create_flow_tree("Test Flow", "test-flow-id")
|
||||
|
||||
if flow_tree:
|
||||
flow_str = str(flow_tree.label)
|
||||
assert "[FLOW]" in flow_str
|
||||
assert "🌊" not in flow_str
|
||||
|
||||
for child in flow_tree.children:
|
||||
child_str = str(child.label)
|
||||
if "Created" in child_str:
|
||||
assert "[CREATED]" in child_str
|
||||
assert "✨" not in child_str
|
||||
|
||||
@patch.dict(os.environ, {"CREWAI_DISABLE_EMOJIS": "true"})
|
||||
def test_lite_agent_without_emojis(self):
|
||||
"""Test that lite agent events work without emojis."""
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
|
||||
formatter.handle_lite_agent_execution("Test Agent", "started")
|
||||
|
||||
if formatter.current_lite_agent_branch:
|
||||
agent_str = str(formatter.current_lite_agent_branch.label)
|
||||
assert "[AGENT]" in agent_str
|
||||
assert "🤖" not in agent_str
|
||||
|
||||
def test_backward_compatibility(self):
|
||||
"""Test that the default behavior (emojis enabled) is preserved."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
formatter = ConsoleFormatter(verbose=True)
|
||||
|
||||
assert not formatter.disable_emojis
|
||||
assert formatter._get_icon("✅") == "✅"
|
||||
assert formatter._get_icon("❌") == "❌"
|
||||
assert formatter._get_icon("🚀") == "🚀"
|
||||
|
||||
tree = formatter.create_crew_tree("Test Crew", "test-id")
|
||||
if tree:
|
||||
tree_str = str(tree.label)
|
||||
assert "🚀" in tree_str
|
||||
assert "[CREW]" not in tree_str
|
||||
Reference in New Issue
Block a user