andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-05-02 15:52:34 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

refactor: use try/finally for event pairing in tool usage

Browse Source
This commit is contained in:
Greyson LaLonde
2026-01-20 01:15:05 -05:00
parent 161f9bd063
commit da07bd4d9f
1 changed files with 283 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
lib/crewai/src/crewai/tools/tool_usage.py
View File

@@ -241,6 +241,9 @@ class ToolUsage:
if self.task: if self.task:
self.task.increment_tools_errors() self.task.increment_tools_errors()
started_at = time.time()
started_event_emitted = False
if self.agent: if self.agent:
event_data = { event_data = {
"agent_key": self.agent.key, "agent_key": self.agent.key,
@@ -258,11 +261,14 @@ class ToolUsage:
event_data["task_name"] = self.task.name or self.task.description event_data["task_name"] = self.task.name or self.task.description
event_data["task_id"] = str(self.task.id) event_data["task_id"] = str(self.task.id)
crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data)) crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data))
started_event_emitted = True
started_at = time.time()
from_cache = False from_cache = False
result = None # type: ignore result = None # type: ignore
should_retry = False
available_tool = None
try:
if self.tools_handler and self.tools_handler.cache: if self.tools_handler and self.tools_handler.cache:
input_str = "" input_str = ""
if calling.arguments: if calling.arguments:
@@ -287,22 +293,20 @@ class ToolUsage:
usage_limit_error = self._check_usage_limit(available_tool, tool.name) usage_limit_error = self._check_usage_limit(available_tool, tool.name)
if usage_limit_error: if usage_limit_error:
try:
result = usage_limit_error result = usage_limit_error
self._telemetry.tool_usage_error(llm=self.function_calling_llm) self._telemetry.tool_usage_error(llm=self.function_calling_llm)
return self._format_result(result=result) result = self._format_result(result=result)
except Exception: # Don't return early - fall through to finally block
if self.task: elif result is None:
self.task.increment_tools_errors()
if result is None:
try: try:
if calling.tool_name in [ if calling.tool_name in [
"Delegate work to coworker", "Delegate work to coworker",
"Ask question to coworker", "Ask question to coworker",
]: ]:
coworker = ( coworker = (
calling.arguments.get("coworker") if calling.arguments else None calling.arguments.get("coworker")
if calling.arguments
else None
) )
if self.task: if self.task:
self.task.increment_delegations(coworker) self.task.increment_delegations(coworker)
@@ -326,28 +330,6 @@ class ToolUsage:
else: else:
arguments = self._add_fingerprint_metadata({}) arguments = self._add_fingerprint_metadata({})
result = await tool.ainvoke(input=arguments) result = await tool.ainvoke(input=arguments)
except Exception as e:
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
self._run_attempts += 1
if self._run_attempts > self._max_parsing_attempts:
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
error_message = self._i18n.errors("tool_usage_exception").format(
error=e, tool=tool.name, tool_inputs=tool.description
)
error = ToolUsageError(
f"\n{error_message}.\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
).message
if self.task:
self.task.increment_tools_errors()
if self.agent and self.agent.verbose:
self._printer.print(
content=f"\n\n{error_message}\n", color="red"
)
return error
if self.task:
self.task.increment_tools_errors()
return await self.ause(calling=calling, tool_string=tool_string)
if self.tools_handler: if self.tools_handler:
should_cache = True should_cache = True
@@ -375,25 +357,19 @@ class ToolUsage:
"tool_args": calling.arguments, "tool_args": calling.arguments,
} }
self.on_tool_use_finished(
tool=tool,
tool_calling=calling,
from_cache=from_cache,
started_at=started_at,
result=result,
)
if ( if (
hasattr(available_tool, "result_as_answer") hasattr(available_tool, "result_as_answer")
and available_tool.result_as_answer # type: ignore and available_tool.result_as_answer
): ):
result_as_answer = available_tool.result_as_answer # type: ignore result_as_answer = available_tool.result_as_answer
data["result_as_answer"] = result_as_answer # type: ignore data["result_as_answer"] = result_as_answer
if self.agent and hasattr(self.agent, "tools_results"): if self.agent and hasattr(self.agent, "tools_results"):
self.agent.tools_results.append(data) self.agent.tools_results.append(data)
if available_tool and hasattr(available_tool, "current_usage_count"): if available_tool and hasattr(
available_tool, "current_usage_count"
):
available_tool.current_usage_count += 1 available_tool.current_usage_count += 1
if ( if (
hasattr(available_tool, "max_usage_count") hasattr(available_tool, "max_usage_count")
@@ -404,6 +380,42 @@ class ToolUsage:
color="blue", color="blue",
) )
except Exception as e:
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
self._run_attempts += 1
if self._run_attempts > self._max_parsing_attempts:
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
error_message = self._i18n.errors(
"tool_usage_exception"
).format(error=e, tool=tool.name, tool_inputs=tool.description)
result = ToolUsageError(
f"\n{error_message}.\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
).message
if self.task:
self.task.increment_tools_errors()
if self.agent and self.agent.verbose:
self._printer.print(
content=f"\n\n{error_message}\n", color="red"
)
else:
if self.task:
self.task.increment_tools_errors()
should_retry = True
finally:
if started_event_emitted:
self.on_tool_use_finished(
tool=tool,
tool_calling=calling,
from_cache=from_cache,
started_at=started_at,
result=result,
)
# Handle retry after finally block ensures finished event was emitted
if should_retry:
return await self.ause(calling=calling, tool_string=tool_string)
return result return result
def _use( def _use(
@@ -412,6 +424,7 @@ class ToolUsage:
tool: CrewStructuredTool, tool: CrewStructuredTool,
calling: ToolCalling | InstructorToolCalling, calling: ToolCalling | InstructorToolCalling,
) -> str: ) -> str:
# Repeated usage check happens before event emission - safe to return early
if self._check_tool_repeated_usage(calling=calling): if self._check_tool_repeated_usage(calling=calling):
try: try:
result = self._i18n.errors("task_repeated_usage").format( result = self._i18n.errors("task_repeated_usage").format(
@@ -428,6 +441,9 @@ class ToolUsage:
if self.task: if self.task:
self.task.increment_tools_errors() self.task.increment_tools_errors()
started_at = time.time()
started_event_emitted = False
if self.agent: if self.agent:
event_data = { event_data = {
"agent_key": self.agent.key, "agent_key": self.agent.key,
@@ -446,17 +462,18 @@ class ToolUsage:
event_data["task_name"] = self.task.name or self.task.description event_data["task_name"] = self.task.name or self.task.description
event_data["task_id"] = str(self.task.id) event_data["task_id"] = str(self.task.id)
crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data)) crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data))
started_event_emitted = True
started_at = time.time()
from_cache = False from_cache = False
result = None # type: ignore result = None # type: ignore
should_retry = False
available_tool = None
try:
if self.tools_handler and self.tools_handler.cache: if self.tools_handler and self.tools_handler.cache:
input_str = "" input_str = ""
if calling.arguments: if calling.arguments:
if isinstance(calling.arguments, dict): if isinstance(calling.arguments, dict):
import json
input_str = json.dumps(calling.arguments) input_str = json.dumps(calling.arguments)
else: else:
input_str = str(calling.arguments) input_str = str(calling.arguments)
@@ -477,22 +494,20 @@ class ToolUsage:
usage_limit_error = self._check_usage_limit(available_tool, tool.name) usage_limit_error = self._check_usage_limit(available_tool, tool.name)
if usage_limit_error: if usage_limit_error:
try:
result = usage_limit_error result = usage_limit_error
self._telemetry.tool_usage_error(llm=self.function_calling_llm) self._telemetry.tool_usage_error(llm=self.function_calling_llm)
return self._format_result(result=result) result = self._format_result(result=result)
except Exception: # Don't return early - fall through to finally block
if self.task: elif result is None:
self.task.increment_tools_errors()
if result is None:
try: try:
if calling.tool_name in [ if calling.tool_name in [
"Delegate work to coworker", "Delegate work to coworker",
"Ask question to coworker", "Ask question to coworker",
]: ]:
coworker = ( coworker = (
calling.arguments.get("coworker") if calling.arguments else None calling.arguments.get("coworker")
if calling.arguments
else None
) )
if self.task: if self.task:
self.task.increment_delegations(coworker) self.task.increment_delegations(coworker)
@@ -507,40 +522,15 @@ class ToolUsage:
for k, v in calling.arguments.items() for k, v in calling.arguments.items()
if k in acceptable_args if k in acceptable_args
} }
# Add fingerprint metadata if available
arguments = self._add_fingerprint_metadata(arguments) arguments = self._add_fingerprint_metadata(arguments)
result = tool.invoke(input=arguments) result = tool.invoke(input=arguments)
except Exception: except Exception:
arguments = calling.arguments arguments = calling.arguments
# Add fingerprint metadata if available
arguments = self._add_fingerprint_metadata(arguments) arguments = self._add_fingerprint_metadata(arguments)
result = tool.invoke(input=arguments) result = tool.invoke(input=arguments)
else: else:
# Add fingerprint metadata even to empty arguments
arguments = self._add_fingerprint_metadata({}) arguments = self._add_fingerprint_metadata({})
result = tool.invoke(input=arguments) result = tool.invoke(input=arguments)
except Exception as e:
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
self._run_attempts += 1
if self._run_attempts > self._max_parsing_attempts:
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
error_message = self._i18n.errors("tool_usage_exception").format(
error=e, tool=tool.name, tool_inputs=tool.description
)
error = ToolUsageError(
f"\n{error_message}.\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
).message
if self.task:
self.task.increment_tools_errors()
if self.agent and self.agent.verbose:
self._printer.print(
content=f"\n\n{error_message}\n", color="red"
)
return error
if self.task:
self.task.increment_tools_errors()
return self.use(calling=calling, tool_string=tool_string)
if self.tools_handler: if self.tools_handler:
should_cache = True should_cache = True
@@ -555,6 +545,7 @@ class ToolUsage:
self.tools_handler.on_tool_use( self.tools_handler.on_tool_use(
calling=calling, output=result, should_cache=should_cache calling=calling, output=result, should_cache=should_cache
) )
self._telemetry.tool_usage( self._telemetry.tool_usage(
llm=self.function_calling_llm, llm=self.function_calling_llm,
tool_name=tool.name, tool_name=tool.name,
@@ -567,25 +558,19 @@ class ToolUsage:
"tool_args": calling.arguments, "tool_args": calling.arguments,
} }
self.on_tool_use_finished(
tool=tool,
tool_calling=calling,
from_cache=from_cache,
started_at=started_at,
result=result,
)
if ( if (
hasattr(available_tool, "result_as_answer") hasattr(available_tool, "result_as_answer")
and available_tool.result_as_answer # type: ignore # Item "None" of "Any | None" has no attribute "cache_function" and available_tool.result_as_answer
): ):
result_as_answer = available_tool.result_as_answer # type: ignore # Item "None" of "Any | None" has no attribute "result_as_answer" result_as_answer = available_tool.result_as_answer
data["result_as_answer"] = result_as_answer # type: ignore data["result_as_answer"] = result_as_answer
if self.agent and hasattr(self.agent, "tools_results"): if self.agent and hasattr(self.agent, "tools_results"):
self.agent.tools_results.append(data) self.agent.tools_results.append(data)
if available_tool and hasattr(available_tool, "current_usage_count"): if available_tool and hasattr(
available_tool, "current_usage_count"
):
available_tool.current_usage_count += 1 available_tool.current_usage_count += 1
if ( if (
hasattr(available_tool, "max_usage_count") hasattr(available_tool, "max_usage_count")
@@ -596,6 +581,42 @@ class ToolUsage:
color="blue", color="blue",
) )
except Exception as e:
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
self._run_attempts += 1
if self._run_attempts > self._max_parsing_attempts:
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
error_message = self._i18n.errors(
"tool_usage_exception"
).format(error=e, tool=tool.name, tool_inputs=tool.description)
result = ToolUsageError(
f"\n{error_message}.\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
).message
if self.task:
self.task.increment_tools_errors()
if self.agent and self.agent.verbose:
self._printer.print(
content=f"\n\n{error_message}\n", color="red"
)
else:
if self.task:
self.task.increment_tools_errors()
should_retry = True
finally:
if started_event_emitted:
self.on_tool_use_finished(
tool=tool,
tool_calling=calling,
from_cache=from_cache,
started_at=started_at,
result=result,
)
# Handle retry after finally block ensures finished event was emitted
if should_retry:
return self.use(calling=calling, tool_string=tool_string)
return result return result
def _format_result(self, result: Any) -> str: def _format_result(self, result: Any) -> str: