feat: print llm tool usage in console

This commit is contained in:
Lucas Gomide
2025-05-29 15:09:14 -03:00
parent f92137712b
commit a2a6f87a59
2 changed files with 76 additions and 14 deletions

View File

@@ -2,7 +2,7 @@ from io import StringIO
from typing import Any, Dict from typing import Any, Dict
from pydantic import Field, PrivateAttr from pydantic import Field, PrivateAttr
from crewai.llm import LLM
from crewai.task import Task from crewai.task import Task
from crewai.telemetry.telemetry import Telemetry from crewai.telemetry.telemetry import Telemetry
from crewai.utilities import Logger from crewai.utilities import Logger
@@ -283,27 +283,43 @@ class EventListener(BaseEventListener):
@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):
self.formatter.handle_tool_usage_started( if isinstance(source, LLM):
self.formatter.current_agent_branch, self.formatter.handle_llm_tool_usage_started(
event.tool_name, event.tool_name,
)
else:
self.formatter.handle_tool_usage_started(
self.formatter.current_agent_branch,
event.tool_name,
self.formatter.current_crew_tree, self.formatter.current_crew_tree,
) )
@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):
self.formatter.handle_tool_usage_finished( if isinstance(source, LLM):
self.formatter.current_tool_branch, self.formatter.handle_llm_tool_usage_finished(
event.tool_name, event.tool_name,
self.formatter.current_crew_tree, )
) else:
self.formatter.handle_tool_usage_finished(
self.formatter.current_tool_branch,
event.tool_name,
self.formatter.current_crew_tree,
)
@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):
self.formatter.handle_tool_usage_error( if isinstance(source, LLM):
self.formatter.current_tool_branch, self.formatter.handle_llm_tool_usage_error(
event.tool_name, event.tool_name,
event.error, event.error,
self.formatter.current_crew_tree, )
else:
self.formatter.handle_tool_usage_error(
self.formatter.current_tool_branch,
event.tool_name,
event.error,
self.formatter.current_crew_tree,
) )
# ----------- LLM EVENTS ----------- # ----------- LLM EVENTS -----------

View File

@@ -17,6 +17,7 @@ class ConsoleFormatter:
current_lite_agent_branch: Optional[Tree] = None current_lite_agent_branch: Optional[Tree] = None
tool_usage_counts: Dict[str, int] = {} tool_usage_counts: Dict[str, int] = {}
current_reasoning_branch: Optional[Tree] = None # Track reasoning status current_reasoning_branch: Optional[Tree] = None # Track reasoning status
current_llm_tool_tree: Optional[Tree] = None
def __init__(self, verbose: bool = False): def __init__(self, verbose: bool = False):
self.console = Console(width=None) self.console = Console(width=None)
@@ -426,6 +427,51 @@ class ConsoleFormatter:
self.print() self.print()
return method_branch return method_branch
def get_llm_tree(self, tool_name: str):
text = Text()
text.append(f"🔧 Using {tool_name} from LLM available_function", style="yellow")
tree = self.current_flow_tree or self.current_crew_tree
if tree:
tree.add(text)
return tree or Tree(text)
def handle_llm_tool_usage_started(
self,
tool_name: str,
):
tree = self.get_llm_tree(tool_name)
self.add_tree_node(tree, "🔄 Tool Usage Started", "green")
self.print(tree)
self.print()
return tree
def handle_llm_tool_usage_finished(
self,
tool_name: str,
):
tree = self.get_llm_tree(tool_name)
self.add_tree_node(tree, "✅ Tool Usage Completed", "green")
self.print(tree)
self.print()
def handle_llm_tool_usage_error(
self,
tool_name: str,
error: str,
):
tree = self.get_llm_tree(tool_name)
self.add_tree_node(tree, "❌ Tool Usage Failed", "red")
self.print(tree)
self.print()
error_content = self.create_status_content(
"Tool Usage Failed", tool_name, "red", Error=error
)
self.print_panel(error_content, "Tool Error", "red")
def handle_tool_usage_started( def handle_tool_usage_started(
self, self,
agent_branch: Optional[Tree], agent_branch: Optional[Tree],