mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-09 17:05:08 +00:00
76b5f72e81
When a tool raises an error, both ToolUsageErrorEvent and ToolUsageFinishedEvent were being emitted. Since both events pop the event scope stack, this caused the agent scope to be incorrectly popped along with the tool scope.
1052 lines
41 KiB
Python
1052 lines
41 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import datetime
|
|
from difflib import SequenceMatcher
|
|
import json
|
|
from json import JSONDecodeError
|
|
from textwrap import dedent
|
|
import time
|
|
from typing import TYPE_CHECKING, Any, Literal
|
|
|
|
import json5
|
|
from json_repair import repair_json # type: ignore[import-untyped]
|
|
|
|
from crewai.events.event_bus import crewai_event_bus
|
|
from crewai.events.types.tool_usage_events import (
|
|
ToolSelectionErrorEvent,
|
|
ToolUsageErrorEvent,
|
|
ToolUsageFinishedEvent,
|
|
ToolUsageStartedEvent,
|
|
ToolValidateInputErrorEvent,
|
|
)
|
|
from crewai.telemetry.telemetry import Telemetry
|
|
from crewai.tools.structured_tool import CrewStructuredTool
|
|
from crewai.tools.tool_calling import InstructorToolCalling, ToolCalling
|
|
from crewai.utilities.agent_utils import (
|
|
get_tool_names,
|
|
render_text_description_and_args,
|
|
)
|
|
from crewai.utilities.converter import Converter
|
|
from crewai.utilities.i18n import I18N, get_i18n
|
|
from crewai.utilities.printer import Printer
|
|
from crewai.utilities.string_utils import sanitize_tool_name
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from crewai.agents.agent_builder.base_agent import BaseAgent
|
|
from crewai.agents.tools_handler import ToolsHandler
|
|
from crewai.lite_agent import LiteAgent
|
|
from crewai.llm import LLM
|
|
from crewai.task import Task
|
|
|
|
|
|
OPENAI_BIGGER_MODELS: list[
|
|
Literal[
|
|
"gpt-4",
|
|
"gpt-4o",
|
|
"o1-preview",
|
|
"o1-mini",
|
|
"o1",
|
|
"o3",
|
|
"o3-mini",
|
|
]
|
|
] = [
|
|
"gpt-4",
|
|
"gpt-4o",
|
|
"o1-preview",
|
|
"o1-mini",
|
|
"o1",
|
|
"o3",
|
|
"o3-mini",
|
|
]
|
|
|
|
|
|
class ToolUsageError(Exception):
|
|
"""Exception raised for errors in the tool usage."""
|
|
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(self.message)
|
|
|
|
|
|
class ToolUsage:
|
|
"""
|
|
Class that represents the usage of a tool by an agent.
|
|
|
|
Attributes:
|
|
task: Task being executed.
|
|
tools_handler: Tools handler that will manage the tool usage.
|
|
tools: List of tools available for the agent.
|
|
tools_description: Description of the tools available for the agent.
|
|
tools_names: Names of the tools available for the agent.
|
|
function_calling_llm: Language model to be used for the tool usage.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
tools_handler: ToolsHandler | None,
|
|
tools: list[CrewStructuredTool],
|
|
task: Task | None,
|
|
function_calling_llm: LLM,
|
|
agent: BaseAgent | LiteAgent | None = None,
|
|
action: Any = None,
|
|
fingerprint_context: dict[str, str] | None = None,
|
|
) -> None:
|
|
self._i18n: I18N = agent.i18n if agent else get_i18n()
|
|
self._printer: Printer = Printer()
|
|
self._telemetry: Telemetry = Telemetry()
|
|
self._run_attempts: int = 1
|
|
self._max_parsing_attempts: int = 3
|
|
self._remember_format_after_usages: int = 3
|
|
self.agent = agent
|
|
self.tools_description = render_text_description_and_args(tools)
|
|
self.tools_names = get_tool_names(tools)
|
|
self.tools_handler = tools_handler
|
|
self.tools = tools
|
|
self.task = task
|
|
self.action = action
|
|
self.function_calling_llm = function_calling_llm
|
|
self.fingerprint_context = fingerprint_context or {}
|
|
|
|
# Set the maximum parsing attempts for bigger models
|
|
if (
|
|
self.function_calling_llm
|
|
and self.function_calling_llm.model in OPENAI_BIGGER_MODELS
|
|
):
|
|
self._max_parsing_attempts = 2
|
|
self._remember_format_after_usages = 4
|
|
|
|
def parse_tool_calling(
|
|
self, tool_string: str
|
|
) -> ToolCalling | InstructorToolCalling | ToolUsageError:
|
|
"""Parse the tool string and return the tool calling."""
|
|
return self._tool_calling(tool_string)
|
|
|
|
def use(
|
|
self, calling: ToolCalling | InstructorToolCalling, tool_string: str
|
|
) -> str:
|
|
if isinstance(calling, ToolUsageError):
|
|
error = calling.message
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
return error
|
|
|
|
try:
|
|
tool = self._select_tool(calling.tool_name)
|
|
except Exception as e:
|
|
error = getattr(e, "message", str(e))
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
return error
|
|
|
|
if (
|
|
isinstance(tool, CrewStructuredTool)
|
|
and sanitize_tool_name(tool.name)
|
|
== sanitize_tool_name(self._i18n.tools("add_image")["name"]) # type: ignore
|
|
):
|
|
try:
|
|
return self._use(tool_string=tool_string, tool=tool, calling=calling)
|
|
|
|
except Exception as e:
|
|
error = getattr(e, "message", str(e))
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
return error
|
|
|
|
return f"{self._use(tool_string=tool_string, tool=tool, calling=calling)}"
|
|
|
|
async def ause(
|
|
self, calling: ToolCalling | InstructorToolCalling, tool_string: str
|
|
) -> str:
|
|
"""Execute a tool asynchronously.
|
|
|
|
Args:
|
|
calling: The tool calling information.
|
|
tool_string: The raw tool string from the agent.
|
|
|
|
Returns:
|
|
The result of the tool execution as a string.
|
|
"""
|
|
if isinstance(calling, ToolUsageError):
|
|
error = calling.message
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
return error
|
|
|
|
try:
|
|
tool = self._select_tool(calling.tool_name)
|
|
except Exception as e:
|
|
error = getattr(e, "message", str(e))
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
return error
|
|
|
|
if (
|
|
isinstance(tool, CrewStructuredTool)
|
|
and sanitize_tool_name(tool.name)
|
|
== sanitize_tool_name(self._i18n.tools("add_image")["name"]) # type: ignore
|
|
):
|
|
try:
|
|
return await self._ause(
|
|
tool_string=tool_string, tool=tool, calling=calling
|
|
)
|
|
except Exception as e:
|
|
error = getattr(e, "message", str(e))
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{error}\n", color="red")
|
|
return error
|
|
|
|
return (
|
|
f"{await self._ause(tool_string=tool_string, tool=tool, calling=calling)}"
|
|
)
|
|
|
|
async def _ause(
|
|
self,
|
|
tool_string: str,
|
|
tool: CrewStructuredTool,
|
|
calling: ToolCalling | InstructorToolCalling,
|
|
) -> str:
|
|
"""Internal async tool execution implementation.
|
|
|
|
Args:
|
|
tool_string: The raw tool string from the agent.
|
|
tool: The tool to execute.
|
|
calling: The tool calling information.
|
|
|
|
Returns:
|
|
The result of the tool execution as a string.
|
|
"""
|
|
if self._check_tool_repeated_usage(calling=calling):
|
|
try:
|
|
result = self._i18n.errors("task_repeated_usage").format(
|
|
tool_names=self.tools_names
|
|
)
|
|
self._telemetry.tool_repeated_usage(
|
|
llm=self.function_calling_llm,
|
|
tool_name=sanitize_tool_name(tool.name),
|
|
attempts=self._run_attempts,
|
|
)
|
|
return self._format_result(result=result)
|
|
except Exception:
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
|
|
started_at = time.time()
|
|
started_event_emitted = False
|
|
|
|
if self.agent:
|
|
event_data = {
|
|
"agent_key": self.agent.key,
|
|
"agent_role": self.agent.role,
|
|
"tool_name": self.action.tool,
|
|
"tool_args": self.action.tool_input,
|
|
"tool_class": self.action.tool,
|
|
"agent": self.agent,
|
|
"run_attempts": self._run_attempts,
|
|
}
|
|
|
|
if self.agent.fingerprint: # type: ignore
|
|
event_data.update(self.agent.fingerprint) # type: ignore
|
|
if self.task:
|
|
event_data["task_name"] = self.task.name or self.task.description
|
|
event_data["task_id"] = str(self.task.id)
|
|
crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data))
|
|
started_event_emitted = True
|
|
|
|
from_cache = False
|
|
result = None # type: ignore
|
|
should_retry = False
|
|
available_tool = None
|
|
error_event_emitted = False
|
|
|
|
try:
|
|
if self.tools_handler and self.tools_handler.cache:
|
|
input_str = ""
|
|
if calling.arguments:
|
|
if isinstance(calling.arguments, dict):
|
|
input_str = json.dumps(calling.arguments)
|
|
else:
|
|
input_str = str(calling.arguments)
|
|
|
|
result = self.tools_handler.cache.read(
|
|
tool=sanitize_tool_name(calling.tool_name), input=input_str
|
|
) # type: ignore
|
|
from_cache = result is not None
|
|
|
|
available_tool = next(
|
|
(
|
|
available_tool
|
|
for available_tool in self.tools
|
|
if sanitize_tool_name(available_tool.name)
|
|
== sanitize_tool_name(tool.name)
|
|
),
|
|
None,
|
|
)
|
|
|
|
usage_limit_error = self._check_usage_limit(
|
|
available_tool, sanitize_tool_name(tool.name)
|
|
)
|
|
if usage_limit_error:
|
|
result = usage_limit_error
|
|
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
|
|
result = self._format_result(result=result)
|
|
# Don't return early - fall through to finally block
|
|
elif result is None:
|
|
try:
|
|
if sanitize_tool_name(calling.tool_name) in [
|
|
sanitize_tool_name("Delegate work to coworker"),
|
|
sanitize_tool_name("Ask question to coworker"),
|
|
]:
|
|
coworker = (
|
|
calling.arguments.get("coworker")
|
|
if calling.arguments
|
|
else None
|
|
)
|
|
if self.task:
|
|
self.task.increment_delegations(coworker)
|
|
|
|
if calling.arguments:
|
|
try:
|
|
acceptable_args = tool.args_schema.model_json_schema()[
|
|
"properties"
|
|
].keys()
|
|
arguments = {
|
|
k: v
|
|
for k, v in calling.arguments.items()
|
|
if k in acceptable_args
|
|
}
|
|
arguments = self._add_fingerprint_metadata(arguments)
|
|
result = await tool.ainvoke(input=arguments)
|
|
except Exception:
|
|
arguments = calling.arguments
|
|
arguments = self._add_fingerprint_metadata(arguments)
|
|
result = await tool.ainvoke(input=arguments)
|
|
else:
|
|
arguments = self._add_fingerprint_metadata({})
|
|
result = await tool.ainvoke(input=arguments)
|
|
|
|
if self.tools_handler:
|
|
should_cache = True
|
|
# Check cache_function on original tool (for tools converted via to_structured_tool)
|
|
original_tool = getattr(available_tool, "_original_tool", None)
|
|
cache_func = None
|
|
if original_tool and hasattr(original_tool, "cache_function"):
|
|
cache_func = original_tool.cache_function
|
|
elif hasattr(available_tool, "cache_function"):
|
|
cache_func = available_tool.cache_function
|
|
|
|
if cache_func:
|
|
should_cache = cache_func(calling.arguments, result)
|
|
|
|
self.tools_handler.on_tool_use(
|
|
calling=calling, output=result, should_cache=should_cache
|
|
)
|
|
|
|
self._telemetry.tool_usage(
|
|
llm=self.function_calling_llm,
|
|
tool_name=sanitize_tool_name(tool.name),
|
|
attempts=self._run_attempts,
|
|
)
|
|
result = self._format_result(result=result)
|
|
data = {
|
|
"result": result,
|
|
"tool_name": sanitize_tool_name(tool.name),
|
|
"tool_args": calling.arguments,
|
|
}
|
|
|
|
if (
|
|
hasattr(available_tool, "result_as_answer")
|
|
and available_tool.result_as_answer
|
|
):
|
|
result_as_answer = available_tool.result_as_answer
|
|
data["result_as_answer"] = result_as_answer
|
|
|
|
if self.agent and hasattr(self.agent, "tools_results"):
|
|
self.agent.tools_results.append(data)
|
|
|
|
if available_tool and hasattr(
|
|
available_tool, "_increment_usage_count"
|
|
):
|
|
# Use _increment_usage_count to sync count to original tool
|
|
available_tool._increment_usage_count()
|
|
if (
|
|
hasattr(available_tool, "max_usage_count")
|
|
and available_tool.max_usage_count is not None
|
|
and self.agent
|
|
and self.agent.verbose
|
|
):
|
|
self._printer.print(
|
|
content=f"Tool '{sanitize_tool_name(available_tool.name)}' usage: {available_tool.current_usage_count}/{available_tool.max_usage_count}",
|
|
color="blue",
|
|
)
|
|
elif available_tool and hasattr(
|
|
available_tool, "current_usage_count"
|
|
):
|
|
available_tool.current_usage_count += 1
|
|
if (
|
|
hasattr(available_tool, "max_usage_count")
|
|
and available_tool.max_usage_count is not None
|
|
and self.agent
|
|
and self.agent.verbose
|
|
):
|
|
self._printer.print(
|
|
content=f"Tool '{sanitize_tool_name(available_tool.name)}' usage: {available_tool.current_usage_count}/{available_tool.max_usage_count}",
|
|
color="blue",
|
|
)
|
|
|
|
except Exception as e:
|
|
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
|
|
error_event_emitted = True
|
|
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=sanitize_tool_name(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
|
|
else:
|
|
result = self._format_result(result=result)
|
|
|
|
finally:
|
|
if started_event_emitted and not error_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
|
|
|
|
def _use(
|
|
self,
|
|
tool_string: str,
|
|
tool: CrewStructuredTool,
|
|
calling: ToolCalling | InstructorToolCalling,
|
|
) -> str:
|
|
# Repeated usage check happens before event emission - safe to return early
|
|
if self._check_tool_repeated_usage(calling=calling):
|
|
try:
|
|
result = self._i18n.errors("task_repeated_usage").format(
|
|
tool_names=self.tools_names
|
|
)
|
|
self._telemetry.tool_repeated_usage(
|
|
llm=self.function_calling_llm,
|
|
tool_name=sanitize_tool_name(tool.name),
|
|
attempts=self._run_attempts,
|
|
)
|
|
return self._format_result(result=result)
|
|
|
|
except Exception:
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
|
|
started_at = time.time()
|
|
started_event_emitted = False
|
|
|
|
if self.agent:
|
|
event_data = {
|
|
"agent_key": self.agent.key,
|
|
"agent_role": self.agent.role,
|
|
"tool_name": self.action.tool,
|
|
"tool_args": self.action.tool_input,
|
|
"tool_class": self.action.tool,
|
|
"agent": self.agent,
|
|
"run_attempts": self._run_attempts,
|
|
}
|
|
|
|
# TODO: Investigate fingerprint attribute availability on BaseAgent/LiteAgent
|
|
if self.agent.fingerprint: # type: ignore
|
|
event_data.update(self.agent.fingerprint) # type: ignore
|
|
if self.task:
|
|
event_data["task_name"] = self.task.name or self.task.description
|
|
event_data["task_id"] = str(self.task.id)
|
|
crewai_event_bus.emit(self, ToolUsageStartedEvent(**event_data))
|
|
started_event_emitted = True
|
|
|
|
from_cache = False
|
|
result = None # type: ignore
|
|
should_retry = False
|
|
available_tool = None
|
|
error_event_emitted = False
|
|
|
|
try:
|
|
if self.tools_handler and self.tools_handler.cache:
|
|
input_str = ""
|
|
if calling.arguments:
|
|
if isinstance(calling.arguments, dict):
|
|
input_str = json.dumps(calling.arguments)
|
|
else:
|
|
input_str = str(calling.arguments)
|
|
|
|
result = self.tools_handler.cache.read(
|
|
tool=sanitize_tool_name(calling.tool_name), input=input_str
|
|
) # type: ignore
|
|
from_cache = result is not None
|
|
|
|
available_tool = next(
|
|
(
|
|
available_tool
|
|
for available_tool in self.tools
|
|
if sanitize_tool_name(available_tool.name)
|
|
== sanitize_tool_name(tool.name)
|
|
),
|
|
None,
|
|
)
|
|
|
|
usage_limit_error = self._check_usage_limit(
|
|
available_tool, sanitize_tool_name(tool.name)
|
|
)
|
|
if usage_limit_error:
|
|
result = usage_limit_error
|
|
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
|
|
result = self._format_result(result=result)
|
|
# Don't return early - fall through to finally block
|
|
elif result is None:
|
|
try:
|
|
if sanitize_tool_name(calling.tool_name) in [
|
|
sanitize_tool_name("Delegate work to coworker"),
|
|
sanitize_tool_name("Ask question to coworker"),
|
|
]:
|
|
coworker = (
|
|
calling.arguments.get("coworker")
|
|
if calling.arguments
|
|
else None
|
|
)
|
|
if self.task:
|
|
self.task.increment_delegations(coworker)
|
|
|
|
if calling.arguments:
|
|
try:
|
|
acceptable_args = tool.args_schema.model_json_schema()[
|
|
"properties"
|
|
].keys()
|
|
arguments = {
|
|
k: v
|
|
for k, v in calling.arguments.items()
|
|
if k in acceptable_args
|
|
}
|
|
arguments = self._add_fingerprint_metadata(arguments)
|
|
result = tool.invoke(input=arguments)
|
|
except Exception:
|
|
arguments = calling.arguments
|
|
arguments = self._add_fingerprint_metadata(arguments)
|
|
result = tool.invoke(input=arguments)
|
|
else:
|
|
arguments = self._add_fingerprint_metadata({})
|
|
result = tool.invoke(input=arguments)
|
|
|
|
if self.tools_handler:
|
|
should_cache = True
|
|
# Check cache_function on original tool (for tools converted via to_structured_tool)
|
|
original_tool = getattr(available_tool, "_original_tool", None)
|
|
cache_func = None
|
|
if original_tool and hasattr(original_tool, "cache_function"):
|
|
cache_func = original_tool.cache_function
|
|
elif hasattr(available_tool, "cache_function"):
|
|
cache_func = available_tool.cache_function
|
|
|
|
if cache_func:
|
|
should_cache = cache_func(calling.arguments, result)
|
|
|
|
self.tools_handler.on_tool_use(
|
|
calling=calling, output=result, should_cache=should_cache
|
|
)
|
|
|
|
self._telemetry.tool_usage(
|
|
llm=self.function_calling_llm,
|
|
tool_name=sanitize_tool_name(tool.name),
|
|
attempts=self._run_attempts,
|
|
)
|
|
result = self._format_result(result=result)
|
|
data = {
|
|
"result": result,
|
|
"tool_name": sanitize_tool_name(tool.name),
|
|
"tool_args": calling.arguments,
|
|
}
|
|
|
|
if (
|
|
hasattr(available_tool, "result_as_answer")
|
|
and available_tool.result_as_answer
|
|
):
|
|
result_as_answer = available_tool.result_as_answer
|
|
data["result_as_answer"] = result_as_answer
|
|
|
|
if self.agent and hasattr(self.agent, "tools_results"):
|
|
self.agent.tools_results.append(data)
|
|
|
|
if available_tool and hasattr(
|
|
available_tool, "_increment_usage_count"
|
|
):
|
|
# Use _increment_usage_count to sync count to original tool
|
|
available_tool._increment_usage_count()
|
|
if (
|
|
hasattr(available_tool, "max_usage_count")
|
|
and available_tool.max_usage_count is not None
|
|
and self.agent
|
|
and self.agent.verbose
|
|
):
|
|
self._printer.print(
|
|
content=f"Tool '{sanitize_tool_name(available_tool.name)}' usage: {available_tool.current_usage_count}/{available_tool.max_usage_count}",
|
|
color="blue",
|
|
)
|
|
elif available_tool and hasattr(
|
|
available_tool, "current_usage_count"
|
|
):
|
|
available_tool.current_usage_count += 1
|
|
if (
|
|
hasattr(available_tool, "max_usage_count")
|
|
and available_tool.max_usage_count is not None
|
|
and self.agent
|
|
and self.agent.verbose
|
|
):
|
|
self._printer.print(
|
|
content=f"Tool '{sanitize_tool_name(available_tool.name)}' usage: {available_tool.current_usage_count}/{available_tool.max_usage_count}",
|
|
color="blue",
|
|
)
|
|
|
|
except Exception as e:
|
|
self.on_tool_error(tool=tool, tool_calling=calling, e=e)
|
|
error_event_emitted = True
|
|
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=sanitize_tool_name(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
|
|
else:
|
|
result = self._format_result(result=result)
|
|
|
|
finally:
|
|
if started_event_emitted and not error_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
|
|
|
|
def _format_result(self, result: Any) -> str:
|
|
if self.task:
|
|
self.task.used_tools += 1
|
|
if self._should_remember_format():
|
|
result = self._remember_format(result=result)
|
|
return str(result)
|
|
|
|
def _should_remember_format(self) -> bool:
|
|
if self.task:
|
|
return self.task.used_tools % self._remember_format_after_usages == 0
|
|
return False
|
|
|
|
def _remember_format(self, result: str) -> str:
|
|
result = str(result)
|
|
result += "\n\n" + self._i18n.slice("tools").format(
|
|
tools=self.tools_description, tool_names=self.tools_names
|
|
)
|
|
return result
|
|
|
|
def _check_tool_repeated_usage(
|
|
self, calling: ToolCalling | InstructorToolCalling
|
|
) -> bool:
|
|
if not self.tools_handler:
|
|
return False
|
|
if last_tool_usage := self.tools_handler.last_used_tool:
|
|
return (
|
|
sanitize_tool_name(calling.tool_name)
|
|
== sanitize_tool_name(last_tool_usage.tool_name)
|
|
) and (calling.arguments == last_tool_usage.arguments)
|
|
return False
|
|
|
|
@staticmethod
|
|
def _check_usage_limit(tool: Any, tool_name: str) -> str | None:
|
|
"""Check if tool has reached its usage limit.
|
|
|
|
Args:
|
|
tool: The tool to check
|
|
tool_name: The name of the tool (used for error message)
|
|
|
|
Returns:
|
|
Error message if limit reached, None otherwise
|
|
"""
|
|
if (
|
|
hasattr(tool, "max_usage_count")
|
|
and tool.max_usage_count is not None
|
|
and tool.current_usage_count >= tool.max_usage_count
|
|
):
|
|
return f"Tool '{tool_name}' has reached its usage limit of {tool.max_usage_count} times and cannot be used anymore."
|
|
return None
|
|
|
|
def _select_tool(self, tool_name: str) -> Any:
|
|
sanitized_input = sanitize_tool_name(tool_name)
|
|
order_tools = sorted(
|
|
self.tools,
|
|
key=lambda tool: SequenceMatcher(
|
|
None, sanitize_tool_name(tool.name), sanitized_input
|
|
).ratio(),
|
|
reverse=True,
|
|
)
|
|
for tool in order_tools:
|
|
sanitized_tool = sanitize_tool_name(tool.name)
|
|
if (
|
|
sanitized_tool == sanitized_input
|
|
or SequenceMatcher(None, sanitized_tool, sanitized_input).ratio() > 0.85
|
|
):
|
|
return tool
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
tool_selection_data: dict[str, Any] = {
|
|
"agent_key": getattr(self.agent, "key", None) if self.agent else None,
|
|
"agent_role": getattr(self.agent, "role", None) if self.agent else None,
|
|
"tool_name": tool_name,
|
|
"tool_args": {},
|
|
"tool_class": self.tools_description,
|
|
}
|
|
if tool_name and tool_name != "":
|
|
error = f"Action '{tool_name}' don't exist, these are the only available Actions:\n{self.tools_description}"
|
|
crewai_event_bus.emit(
|
|
self,
|
|
ToolSelectionErrorEvent(
|
|
**tool_selection_data,
|
|
error=error,
|
|
),
|
|
)
|
|
raise Exception(error)
|
|
error = f"I forgot the Action name, these are the only available Actions: {self.tools_description}"
|
|
crewai_event_bus.emit(
|
|
self,
|
|
ToolSelectionErrorEvent(
|
|
**tool_selection_data,
|
|
error=error,
|
|
),
|
|
)
|
|
raise Exception(error)
|
|
|
|
def _render(self) -> str:
|
|
"""Render the tool name and description in plain text."""
|
|
descriptions = [tool.description for tool in self.tools]
|
|
return "\n--\n".join(descriptions)
|
|
|
|
def _function_calling(
|
|
self, tool_string: str
|
|
) -> ToolCalling | InstructorToolCalling:
|
|
model = (
|
|
InstructorToolCalling
|
|
if self.function_calling_llm.supports_function_calling()
|
|
else ToolCalling
|
|
)
|
|
converter = Converter(
|
|
text=f"Only tools available:\n###\n{self._render()}\n\nReturn a valid schema for the tool, the tool name must be exactly equal one of the options, use this text to inform the valid output schema:\n\n### TEXT \n{tool_string}",
|
|
llm=self.function_calling_llm,
|
|
model=model,
|
|
instructions=dedent(
|
|
"""\
|
|
The schema should have the following structure, only two keys:
|
|
- tool_name: str
|
|
- arguments: dict (always a dictionary, with all arguments being passed)
|
|
|
|
Example:
|
|
{"tool_name": "tool name", "arguments": {"arg_name1": "value", "arg_name2": 2}}""",
|
|
),
|
|
max_attempts=1,
|
|
)
|
|
tool_object = converter.to_pydantic()
|
|
if not isinstance(tool_object, (ToolCalling, InstructorToolCalling)):
|
|
raise ToolUsageError("Failed to parse tool calling")
|
|
|
|
return tool_object
|
|
|
|
def _original_tool_calling(
|
|
self, tool_string: str, raise_error: bool = False
|
|
) -> ToolCalling | InstructorToolCalling | ToolUsageError:
|
|
tool_name = self.action.tool
|
|
tool = self._select_tool(tool_name)
|
|
try:
|
|
arguments = self._validate_tool_input(self.action.tool_input)
|
|
|
|
except Exception:
|
|
if raise_error:
|
|
raise
|
|
return ToolUsageError(f"{self._i18n.errors('tool_arguments_error')}")
|
|
|
|
if not isinstance(arguments, dict):
|
|
if raise_error:
|
|
raise
|
|
return ToolUsageError(f"{self._i18n.errors('tool_arguments_error')}")
|
|
|
|
return ToolCalling(
|
|
tool_name=sanitize_tool_name(tool.name),
|
|
arguments=arguments,
|
|
)
|
|
|
|
def _tool_calling(
|
|
self, tool_string: str
|
|
) -> ToolCalling | InstructorToolCalling | ToolUsageError:
|
|
try:
|
|
try:
|
|
return self._original_tool_calling(tool_string, raise_error=True)
|
|
except Exception:
|
|
if self.function_calling_llm:
|
|
return self._function_calling(tool_string)
|
|
return self._original_tool_calling(tool_string)
|
|
except Exception as e:
|
|
self._run_attempts += 1
|
|
if self._run_attempts > self._max_parsing_attempts:
|
|
self._telemetry.tool_usage_error(llm=self.function_calling_llm)
|
|
if self.task:
|
|
self.task.increment_tools_errors()
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=f"\n\n{e}\n", color="red")
|
|
return ToolUsageError(
|
|
f"{self._i18n.errors('tool_usage_error').format(error=e)}\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
|
|
)
|
|
return self._tool_calling(tool_string)
|
|
|
|
def _validate_tool_input(self, tool_input: str | None) -> dict[str, Any]:
|
|
if tool_input is None:
|
|
return {}
|
|
|
|
if not isinstance(tool_input, str) or not tool_input.strip():
|
|
raise Exception(
|
|
"Tool input must be a valid dictionary in JSON or Python literal format"
|
|
)
|
|
|
|
# Attempt 1: Parse as JSON
|
|
try:
|
|
arguments = json.loads(tool_input)
|
|
if isinstance(arguments, dict):
|
|
return arguments
|
|
except (JSONDecodeError, TypeError):
|
|
pass # Continue to the next parsing attempt
|
|
|
|
# Attempt 2: Parse as Python literal
|
|
try:
|
|
arguments = ast.literal_eval(tool_input)
|
|
if isinstance(arguments, dict):
|
|
return arguments
|
|
except (ValueError, SyntaxError):
|
|
repaired_input = repair_json(tool_input)
|
|
# Continue to the next parsing attempt
|
|
|
|
# Attempt 3: Parse as JSON5
|
|
try:
|
|
arguments = json5.loads(tool_input)
|
|
if isinstance(arguments, dict):
|
|
return arguments
|
|
except (JSONDecodeError, ValueError, TypeError):
|
|
pass # Continue to the next parsing attempt
|
|
|
|
# Attempt 4: Repair JSON
|
|
try:
|
|
repaired_input = str(repair_json(tool_input, skip_json_loads=True))
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(
|
|
content=f"Repaired JSON: {repaired_input}", color="blue"
|
|
)
|
|
arguments = json.loads(repaired_input)
|
|
if isinstance(arguments, dict):
|
|
return arguments
|
|
except Exception as e:
|
|
error = f"Failed to repair JSON: {e}"
|
|
if self.agent and self.agent.verbose:
|
|
self._printer.print(content=error, color="red")
|
|
|
|
error_message = (
|
|
"Tool input must be a valid dictionary in JSON or Python literal format"
|
|
)
|
|
self._emit_validate_input_error(error_message)
|
|
# If all parsing attempts fail, raise an error
|
|
raise Exception(error_message)
|
|
|
|
def _emit_validate_input_error(self, final_error: str) -> None:
|
|
tool_selection_data = {
|
|
"agent_key": getattr(self.agent, "key", None) if self.agent else None,
|
|
"agent_role": getattr(self.agent, "role", None) if self.agent else None,
|
|
"tool_name": self.action.tool,
|
|
"tool_args": str(self.action.tool_input),
|
|
"tool_class": self.__class__.__name__,
|
|
"agent": self.agent, # Adding agent for fingerprint extraction
|
|
}
|
|
|
|
# Include fingerprint context if available
|
|
if self.fingerprint_context:
|
|
tool_selection_data.update(self.fingerprint_context)
|
|
|
|
crewai_event_bus.emit(
|
|
self,
|
|
ToolValidateInputErrorEvent(**tool_selection_data, error=final_error),
|
|
)
|
|
|
|
def on_tool_error(
|
|
self,
|
|
tool: Any,
|
|
tool_calling: ToolCalling | InstructorToolCalling,
|
|
e: Exception,
|
|
) -> None:
|
|
event_data = self._prepare_event_data(tool, tool_calling)
|
|
event_data.update(
|
|
{
|
|
"task_id": str(self.task.id) if self.task else None,
|
|
"task_name": self.task.name or self.task.description
|
|
if self.task
|
|
else None,
|
|
}
|
|
)
|
|
crewai_event_bus.emit(
|
|
self,
|
|
ToolUsageErrorEvent(
|
|
**{
|
|
**event_data,
|
|
"error": e,
|
|
}
|
|
),
|
|
)
|
|
|
|
def on_tool_use_finished(
|
|
self,
|
|
tool: Any,
|
|
tool_calling: ToolCalling | InstructorToolCalling,
|
|
from_cache: bool,
|
|
started_at: float,
|
|
result: Any,
|
|
) -> None:
|
|
finished_at = time.time()
|
|
event_data = self._prepare_event_data(tool, tool_calling)
|
|
event_data.update(
|
|
{
|
|
"started_at": datetime.datetime.fromtimestamp(started_at),
|
|
"finished_at": datetime.datetime.fromtimestamp(finished_at),
|
|
"from_cache": from_cache,
|
|
"output": result,
|
|
}
|
|
)
|
|
if self.task:
|
|
event_data["task_id"] = str(self.task.id)
|
|
event_data["task_name"] = self.task.name or self.task.description
|
|
crewai_event_bus.emit(self, ToolUsageFinishedEvent(**event_data))
|
|
|
|
def _prepare_event_data(
|
|
self, tool: Any, tool_calling: ToolCalling | InstructorToolCalling
|
|
) -> dict[str, Any]:
|
|
event_data = {
|
|
"run_attempts": self._run_attempts,
|
|
"delegations": self.task.delegations if self.task else 0,
|
|
"tool_name": sanitize_tool_name(tool.name),
|
|
"tool_args": tool_calling.arguments,
|
|
"tool_class": tool.__class__.__name__,
|
|
"agent_key": (
|
|
getattr(self.agent, "key", "unknown") if self.agent else "unknown"
|
|
),
|
|
"agent_role": (
|
|
getattr(self.agent, "_original_role", None)
|
|
or getattr(self.agent, "role", "unknown")
|
|
if self.agent
|
|
else "unknown"
|
|
),
|
|
}
|
|
|
|
# Include fingerprint context if available
|
|
if self.fingerprint_context:
|
|
event_data.update(self.fingerprint_context)
|
|
|
|
return event_data
|
|
|
|
def _add_fingerprint_metadata(self, arguments: dict[str, Any]) -> dict[str, Any]:
|
|
"""Add fingerprint metadata to tool arguments if available.
|
|
|
|
Args:
|
|
arguments: The original tool arguments
|
|
|
|
Returns:
|
|
Updated arguments dictionary with fingerprint metadata
|
|
"""
|
|
# Create a shallow copy to avoid modifying the original
|
|
arguments = arguments.copy()
|
|
|
|
# Add security metadata under a designated key
|
|
if "security_context" not in arguments:
|
|
arguments["security_context"] = {}
|
|
|
|
security_context = arguments["security_context"]
|
|
|
|
# Add agent fingerprint if available
|
|
if self.agent and hasattr(self.agent, "security_config"):
|
|
security_config = getattr(self.agent, "security_config", None)
|
|
if security_config and hasattr(security_config, "fingerprint"):
|
|
try:
|
|
security_context["agent_fingerprint"] = (
|
|
security_config.fingerprint.to_dict()
|
|
)
|
|
except AttributeError:
|
|
pass
|
|
|
|
# Add task fingerprint if available
|
|
if self.task and hasattr(self.task, "security_config"):
|
|
security_config = getattr(self.task, "security_config", None)
|
|
if security_config and hasattr(security_config, "fingerprint"):
|
|
try:
|
|
security_context["task_fingerprint"] = (
|
|
security_config.fingerprint.to_dict()
|
|
)
|
|
except AttributeError:
|
|
pass
|
|
|
|
return arguments
|