diff --git a/src/crewai/tools/base_tool.py b/src/crewai/tools/base_tool.py index b3c0f997c..5425d0f35 100644 --- a/src/crewai/tools/base_tool.py +++ b/src/crewai/tools/base_tool.py @@ -37,6 +37,8 @@ class BaseTool(BaseModel, ABC): """Function that will be used to determine if the tool should be cached, should return a boolean. If None, the tool will be cached.""" result_as_answer: bool = False """Flag to check if the tool should be the final agent answer.""" + allow_repeated_usage: bool = False + """Whether the tool permits repeated usage with same arguments.""" @validator("args_schema", always=True, pre=True) def _default_args_schema( diff --git a/src/crewai/tools/tool_usage.py b/src/crewai/tools/tool_usage.py index 25e4b126a..36642b25e 100644 --- a/src/crewai/tools/tool_usage.py +++ b/src/crewai/tools/tool_usage.py @@ -279,6 +279,10 @@ class ToolUsage: if not self.tools_handler: return False # type: ignore # No return value expected if last_tool_usage := self.tools_handler.last_used_tool: + tool = self._select_tool(calling.tool_name) + # If the tool allows repeated usage, don't check arguments + if getattr(tool, "allow_repeated_usage", False): + return False # type: ignore # No return value expected return (calling.tool_name == last_tool_usage.tool_name) and ( # type: ignore # No return value expected calling.arguments == last_tool_usage.arguments ) diff --git a/tests/agent_test.py b/tests/agent_test.py index b5b3aae93..b775b1a42 100644 --- a/tests/agent_test.py +++ b/tests/agent_test.py @@ -546,6 +546,47 @@ def test_agent_moved_on_after_max_iterations(): assert output == "42" +@pytest.mark.vcr(filter_headers=["authorization"]) +def test_agent_repeated_tool_usage_respects_allow_repeated_usage(capsys): + @tool + def repeatable_tool(anything: str) -> float: + """A tool that allows being used repeatedly with the same input.""" + return 42 + + # Patch the tool to set allow_repeated_usage to True + repeatable_tool.allow_repeated_usage = True + + agent = Agent( + role="test role", + goal="test goal", + backstory="test backstory", + max_iter=4, + llm="gpt-4", + allow_delegation=False, + verbose=True, + ) + + task = Task( + description="Use the repeatable tool with the same input multiple times.", + expected_output="The result of using the repeatable tool", + ) + + # force cleaning cache + agent.tools_handler.cache = CacheHandler() + agent.execute_task( + task=task, + tools=[repeatable_tool], + ) + + captured = capsys.readouterr() + + # Should NOT show the repeated usage error + assert ( + "I tried reusing the same input, I must stop using this action input. I'll try something else instead." + not in captured.out + ) + + @pytest.mark.vcr(filter_headers=["authorization"]) def test_agent_respect_the_max_rpm_set(capsys): @tool diff --git a/tests/cassettes/test_agent_repeated_tool_usage_respects_allow_repeated_usage.yaml b/tests/cassettes/test_agent_repeated_tool_usage_respects_allow_repeated_usage.yaml new file mode 100644 index 000000000..df8109a23 --- /dev/null +++ b/tests/cassettes/test_agent_repeated_tool_usage_respects_allow_repeated_usage.yaml @@ -0,0 +1,95 @@ +interactions: +- request: + body: '{"messages": [{"role": "system", "content": "You are test role. test backstory\nYour + personal goal is: test goal\nYou ONLY have access to the following tools, and + should NEVER make up tools that are not listed here:\n\nTool Name: repeatable_tool\nTool + Arguments: {''anything'': {''description'': None, ''type'': ''str''}}\nTool + Description: A tool that allows being used repeatedly with the same input.\n\nIMPORTANT: + Use the following format in your response:\n\n```\nThought: you should always + think about what to do\nAction: the action to take, only one name of [repeatable_tool], + just the name, exactly as it''s written.\nAction Input: the input to the action, + just a simple JSON object, enclosed in curly braces, using \" to wrap keys and + values.\nObservation: the result of the action\n```\n\nOnce all necessary information + is gathered, return the following format:\n\n```\nThought: I now know the final + answer\nFinal Answer: the final answer to the original input question\n```"}, + {"role": "user", "content": "\nCurrent Task: Use the repeatable tool with the + same input multiple times.\n\nThis is the expected criteria for your final answer: + The result of using the repeatable tool\nyou MUST return the actual complete + content as the final answer, not a summary.\n\nBegin! This is VERY important + to you, use the tools available and give your best Final Answer, your job depends + on it!\n\nThought:"}], "model": "gpt-4", "stop": ["\nObservation:"]}' + headers: + accept: + - application/json + accept-encoding: + - gzip, deflate + connection: + - keep-alive + content-length: + - '1443' + content-type: + - application/json + host: + - api.openai.com + user-agent: + - OpenAI/Python 1.61.0 + x-stainless-arch: + - x64 + x-stainless-async: + - 'false' + x-stainless-lang: + - python + x-stainless-os: + - Linux + x-stainless-package-version: + - 1.61.0 + x-stainless-raw-response: + - 'true' + x-stainless-retry-count: + - '0' + x-stainless-runtime: + - CPython + x-stainless-runtime-version: + - 3.12.7 + method: POST + uri: https://api.openai.com/v1/chat/completions + response: + content: "{\n \"error\": {\n \"message\": \"Incorrect API key provided: + sk-proj-********************************************************************************************************************************************************sLcA. + You can find your API key at https://platform.openai.com/account/api-keys.\",\n + \ \"type\": \"invalid_request_error\",\n \"param\": null,\n \"code\": + \"invalid_api_key\"\n }\n}\n" + headers: + CF-RAY: + - 923d7d097e94585c-SEA + Connection: + - keep-alive + Content-Length: + - '414' + Content-Type: + - application/json; charset=utf-8 + Date: + - Fri, 21 Mar 2025 12:35:18 GMT + Server: + - cloudflare + Set-Cookie: + - __cf_bm=Q1SICHUtjtv5VIyjejhOK84VaDt9c0W.OVC6v_gypkQ-1742560518-1.0.1.1-qJY3vQUXlr.VsRsaGGOWSlwiIAp08q3Lt8WnlqIyScSZLPbR0lKV.af50DgwmKKgMtmbt3i27M3b_InDvj8zqEeADyauevHK67qwXvnimEo; + path=/; expires=Fri, 21-Mar-25 13:05:18 GMT; domain=.api.openai.com; HttpOnly; + Secure; SameSite=None + - _cfuvid=BrdkkJU.eT5POa3a8EbLDfLnncr8znx1G52s.PDhIOE-1742560518752-0.0.1.1-604800000; + path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None + X-Content-Type-Options: + - nosniff + alt-svc: + - h3=":443"; ma=86400 + cf-cache-status: + - DYNAMIC + strict-transport-security: + - max-age=31536000; includeSubDomains; preload + vary: + - Origin + x-request-id: + - req_88d97a89c4b41a4e306821c931177663 + http_version: HTTP/1.1 + status_code: 401 +version: 1 diff --git a/tests/tools/test_tool_repeated_usage_allowed.py b/tests/tools/test_tool_repeated_usage_allowed.py new file mode 100644 index 000000000..f7a479a41 --- /dev/null +++ b/tests/tools/test_tool_repeated_usage_allowed.py @@ -0,0 +1,48 @@ +import pytest +from unittest.mock import MagicMock + +from crewai.tools import BaseTool +from crewai.tools.tool_usage import ToolUsage +from crewai.tools.tool_calling import ToolCalling + + +def test_tool_repeated_usage_allowed(): + """Test that a tool with allow_repeated_usage=True can be used repeatedly with same args.""" + + class RepeatedUsageTool(BaseTool): + name: str = "Repeated Usage Tool" + description: str = "A tool that can be used repeatedly with the same arguments" + allow_repeated_usage: bool = True + + def _run(self, test_arg: str) -> str: + return f"Used with arg: {test_arg}" + + # Setup tool usage + tool = RepeatedUsageTool() + tools_handler = MagicMock() + tools_handler.last_used_tool = ToolCalling( + tool_name="Repeated Usage Tool", + arguments={"test_arg": "test"} + ) + + tool_usage = ToolUsage( + tools_handler=tools_handler, + tools=[tool], + original_tools=[tool], + tools_description="Test tools", + tools_names="Repeated Usage Tool", + agent=MagicMock(), + task=MagicMock(), + function_calling_llm=MagicMock(), + action=MagicMock(), + ) + + # Create a new tool calling with the same arguments + calling = ToolCalling( + tool_name="Repeated Usage Tool", + arguments={"test_arg": "test"} + ) + + # This should return False since the tool allows repeated usage + result = tool_usage._check_tool_repeated_usage(calling=calling) + assert result is False