From 74dfcc723a63f1fa4eca351e4024ff7c9c32c87e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 9 Feb 2025 20:17:15 +0000 Subject: [PATCH] refactor: improve delegation validation and testing based on review Co-Authored-By: Joe Moura --- src/crewai/agents/agent_builder/base_agent.py | 33 +++++-- .../tools/agent_tools/base_agent_tools.py | 49 ++++++++-- src/crewai/translations/en.json | 2 +- tests/agents/test_delegation.py | 92 ++++++++++++++----- 4 files changed, 139 insertions(+), 37 deletions(-) diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index 1dcc0287b..64dd84d3f 100644 --- a/src/crewai/agents/agent_builder/base_agent.py +++ b/src/crewai/agents/agent_builder/base_agent.py @@ -110,6 +110,13 @@ class BaseAgent(ABC, BaseModel): allowed_agents: Optional[List[str]] = Field( default=None, description="List of agent roles that this agent is allowed to delegate tasks to.", + docstring=""" + Specifies which agent roles this agent can delegate tasks to. When set: + - Must be a list of role names as strings + - Cannot be empty if delegation is enabled + - Case-insensitive matching is used for role names + - If None, agent can delegate to any other agent (when allow_delegation is True) + """, ) tools: Optional[List[Any]] = Field( default_factory=list, description="Tools at agents' disposal" @@ -178,12 +185,8 @@ class BaseAgent(ABC, BaseModel): f"{field} must be provided either directly or through config" ) - # Validate allowed_agents if delegation is enabled - if self.allow_delegation and self.allowed_agents is not None: - if not isinstance(self.allowed_agents, list): - raise ValueError("allowed_agents must be a list of strings") - if not all(isinstance(agent, str) for agent in self.allowed_agents): - raise ValueError("all entries in allowed_agents must be strings") + # Validate allowed_agents configuration + self._validate_allowed_agents() # Set private attributes self._logger = Logger(verbose=self.verbose) @@ -225,6 +228,24 @@ class BaseAgent(ABC, BaseModel): ] return md5("|".join(source).encode(), usedforsecurity=False).hexdigest() + def _validate_allowed_agents(self) -> None: + """Validate allowed_agents configuration. + + Raises: + ValueError: If allowed_agents is not properly configured: + - Not a list of strings when specified + - Empty list when delegation is enabled + - Contains non-string entries + """ + if self.allow_delegation and self.allowed_agents is not None: + if not isinstance(self.allowed_agents, list): + raise ValueError("allowed_agents must be a list of strings") + if not all(isinstance(agent, str) for agent in self.allowed_agents): + raise ValueError("all entries in allowed_agents must be strings") + if len(self.allowed_agents) == 0: + raise ValueError("allowed_agents cannot be empty when delegation is enabled") + + @abstractmethod def execute_task( self, diff --git a/src/crewai/tools/agent_tools/base_agent_tools.py b/src/crewai/tools/agent_tools/base_agent_tools.py index 85785dbe3..70bd66983 100644 --- a/src/crewai/tools/agent_tools/base_agent_tools.py +++ b/src/crewai/tools/agent_tools/base_agent_tools.py @@ -16,6 +16,39 @@ class BaseAgentTool(BaseTool): default_factory=I18N, description="Internationalization settings" ) + def _get_agent_by_id(self, agent_id: UUID4) -> Optional[BaseAgent]: + """Helper method to find agent by ID.""" + return next((a for a in self.agents if a.id == agent_id), None) + + def _get_agent_by_role(self, role: str) -> Optional[BaseAgent]: + """Helper method to find agent by role (case-insensitive).""" + return next( + (a for a in self.agents if a.role.casefold() == role.casefold()), + None + ) + + def _check_delegation_authorization( + self, delegating_agent: BaseAgent, target_role: str + ) -> Optional[str]: + """Verify if delegation is authorized. + + Args: + delegating_agent: The agent attempting to delegate + target_role: The role of the agent being delegated to + + Returns: + Optional[str]: Error message if delegation is not authorized, None otherwise + """ + if (delegating_agent.allowed_agents is not None and + not any(allowed.casefold() == target_role.casefold() + for allowed in delegating_agent.allowed_agents)): + return self.i18n.errors("agent_tool_unauthorized_delegation").format( + coworker=target_role, + allowed_agents="\n".join([f"- {role}" for role in delegating_agent.allowed_agents]) + ) + return None + + def _get_coworker(self, coworker: Optional[str], **kwargs) -> Optional[str]: coworker = coworker or kwargs.get("co_worker") or kwargs.get("coworker") if coworker: @@ -58,15 +91,17 @@ class BaseAgentTool(BaseTool): ) ) - # Check if delegation is allowed based on allowed_agents list - delegating_agent = [a for a in self.agents if a.id == self.agent_id][0] - if (delegating_agent.allowed_agents is not None and - agent[0].role not in delegating_agent.allowed_agents): - return self.i18n.errors("agent_tool_unauthorized_delegation").format( - coworker=agent[0].role, - allowed_agents="\n".join([f"- {role}" for role in delegating_agent.allowed_agents]) + # Get delegating agent and check authorization + delegating_agent = self._get_agent_by_id(self.agent_id) + if not delegating_agent: + return self.i18n.errors("agent_tool_unexisting_coworker").format( + coworkers="\n".join([f"- {agent.role}" for agent in self.agents]) ) + auth_error = self._check_delegation_authorization(delegating_agent, agent[0].role) + if auth_error: + return auth_error + agent = agent[0] task_with_assigned_agent = Task( # type: ignore # Incompatible types in assignment (expression has type "Task", variable has type "str") description=task, diff --git a/src/crewai/translations/en.json b/src/crewai/translations/en.json index 40bb8b262..fec961ee8 100644 --- a/src/crewai/translations/en.json +++ b/src/crewai/translations/en.json @@ -34,7 +34,7 @@ "tool_arguments_error": "Error: the Action Input is not a valid key, value dictionary.", "wrong_tool_name": "You tried to use the tool {tool}, but it doesn't exist. You must use one of the following tools, use one at time: {tools}.", "tool_usage_exception": "I encountered an error while trying to use the tool. This was the error: {error}.\n Tool {tool} accepts these inputs: {tool_inputs}", - "agent_tool_unauthorized_delegation": "I cannot delegate this task to {coworker} as I am only allowed to delegate to: \n{allowed_agents}" + "agent_tool_unauthorized_delegation": "Authorization Error: Cannot delegate task to {coworker}.\nThis agent is only authorized to delegate to:\n{allowed_agents}\nPlease select an authorized agent for delegation." }, "tools": { "delegate_work": "Delegate a specific task to one of the following coworkers: {coworkers}\nThe input to this tool should be the coworker, the task you want them to do, and ALL necessary context to execute the task, they know nothing about the task, so share absolute everything you know, don't reference things but instead explain them.", diff --git a/tests/agents/test_delegation.py b/tests/agents/test_delegation.py index 322d56cbf..76614784a 100644 --- a/tests/agents/test_delegation.py +++ b/tests/agents/test_delegation.py @@ -56,8 +56,8 @@ def test_delegate_work_with_allowed_agents(): ) # Verify delegation was allowed - assert "error" not in result.lower() - assert "unauthorized" not in result.lower() + assert "authorization error" not in result.lower() + assert "cannot delegate" not in result.lower() def test_delegate_work_with_unauthorized_agent(): """Test failed delegation to unauthorized agent.""" @@ -109,21 +109,66 @@ def test_delegate_work_with_unauthorized_agent(): ) # Verify delegation was blocked with proper error message - assert "cannot delegate this task" in result.lower() + assert "authorization error" in result.lower() assert "tech manager" in result.lower() assert "communications manager" in result.lower() -def test_delegate_work_without_allowed_agents(): - """Test delegation works normally when no allowed_agents is specified.""" +@pytest.mark.parametrize("scenario", [ + { + "name": "empty_allowed_agents", + "delegating_agent": { + "role": "Manager", + "allow_delegation": True, + "allowed_agents": [] + }, + "target_agent": "Worker", + "should_succeed": False, + "error_contains": "cannot be empty" + }, + { + "name": "case_insensitive_match", + "delegating_agent": { + "role": "Manager", + "allow_delegation": True, + "allowed_agents": ["Worker"] + }, + "target_agent": "WORKER", + "should_succeed": True + }, + { + "name": "unauthorized_delegation", + "delegating_agent": { + "role": "Manager", + "allow_delegation": True, + "allowed_agents": ["Worker A"] + }, + "target_agent": "Worker B", + "should_succeed": False, + "error_contains": "Authorization Error" + }, + { + "name": "no_allowed_agents_specified", + "delegating_agent": { + "role": "Manager", + "allow_delegation": True, + "allowed_agents": None + }, + "target_agent": "Worker", + "should_succeed": True + } +]) +def test_delegation_scenarios(scenario): + """Test various delegation scenarios.""" # Create agents - manager = Agent( - role="Manager", + delegating_agent = Agent( + role=scenario["delegating_agent"]["role"], goal="Manage the team", backstory="An experienced manager", - allow_delegation=True # No allowed_agents specified + allow_delegation=scenario["delegating_agent"]["allow_delegation"], + allowed_agents=scenario["delegating_agent"]["allowed_agents"] ) - worker = Agent( - role="Worker", + target_agent = Agent( + role=scenario["target_agent"], goal="Do the work", backstory="A skilled worker", allow_delegation=False @@ -138,29 +183,30 @@ def test_delegate_work_without_allowed_agents(): } }] } - manager.llm = MagicMock() - manager.llm.invoke = MagicMock(return_value=mock_response) - manager.llm.call = MagicMock(return_value=mock_content) - worker.llm = MagicMock() - worker.llm.invoke = MagicMock(return_value=mock_response) - worker.llm.call = MagicMock(return_value=mock_content) + for agent in [delegating_agent, target_agent]: + agent.llm = MagicMock() + agent.llm.invoke = MagicMock(return_value=mock_response) + agent.llm.call = MagicMock(return_value=mock_content) # Create crew and tool - crew = Crew(agents=[manager, worker]) + crew = Crew(agents=[delegating_agent, target_agent]) tool = DelegateWorkTool( name="Delegate work to coworker", description="Tool for delegating work to coworkers", - agents=[manager, worker], - agent_id=manager.id + agents=[delegating_agent, target_agent], + agent_id=delegating_agent.id ) # Test delegation result = tool._execute( - agent_name="Worker", + agent_name=scenario["target_agent"], task="Complete task", context="Important task" ) - # Verify delegation was allowed - assert "error" not in result.lower() - assert "unauthorized" not in result.lower() + # Verify results + if scenario["should_succeed"]: + assert "authorization error" not in result.lower() + assert "cannot delegate" not in result.lower() + else: + assert scenario["error_contains"].lower() in result.lower()