From e06e39899c4c16c706f1cad22ba70f6b1c8734ac Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 30 May 2025 05:55:26 +0000 Subject: [PATCH] feat: add target_agents parameter to allow constrained delegation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: João --- src/crewai/agent.py | 16 +++ tests/test_target_agents_delegation.py | 132 ++++++++++++++++++++ tests/tools/agent_tools/agent_tools_test.py | 45 +++++++ 3 files changed, 193 insertions(+) create mode 100644 tests/test_target_agents_delegation.py diff --git a/src/crewai/agent.py b/src/crewai/agent.py index 9a7373336..fdbb8e6ec 100644 --- a/src/crewai/agent.py +++ b/src/crewai/agent.py @@ -155,6 +155,10 @@ class Agent(BaseAgent): default=None, description="The Agent's role to be used from your repository.", ) + target_agents: Optional[List[str]] = Field( + default=None, + description="List of agent roles that this agent can delegate tasks to. If None, can delegate to all agents.", + ) @model_validator(mode="before") def validate_from_repository(cls, v): @@ -549,6 +553,18 @@ class Agent(BaseAgent): ) def get_delegation_tools(self, agents: List[BaseAgent]): + if hasattr(self, 'target_agents') and self.target_agents is not None: + filtered_agents = [ + agent for agent in agents + if agent.role in self.target_agents + ] + if len(filtered_agents) < len(self.target_agents): + available_roles = [agent.role for agent in agents] + missing_roles = set(self.target_agents) - set(available_roles) + if hasattr(self, '_logger'): + self._logger.log("warning", f"Some target agents not found for delegation: {missing_roles}") + agents = filtered_agents + agent_tools = AgentTools(agents=agents) tools = agent_tools.tools() return tools diff --git a/tests/test_target_agents_delegation.py b/tests/test_target_agents_delegation.py new file mode 100644 index 000000000..acb64bc48 --- /dev/null +++ b/tests/test_target_agents_delegation.py @@ -0,0 +1,132 @@ +"""Test target_agents delegation functionality.""" + +import pytest +from crewai.agent import Agent +from crewai.crew import Crew +from crewai.task import Task +from crewai.tools.agent_tools.agent_tools import AgentTools + +def test_target_agents_filters_delegation_tools(): + """Test that target_agents properly filters available agents for delegation.""" + researcher = Agent( + role="researcher", + goal="Research topics", + backstory="Expert researcher", + allow_delegation=True, + target_agents=["writer"] + ) + + writer = Agent( + role="writer", + goal="Write content", + backstory="Expert writer", + allow_delegation=False + ) + + analyst = Agent( + role="analyst", + goal="Analyze data", + backstory="Expert analyst", + allow_delegation=False + ) + + tools = researcher.get_delegation_tools([writer, analyst]) + delegate_tool = tools[0] + + result = delegate_tool.run( + coworker="writer", + task="Write an article", + context="About AI" + ) + assert "Error executing tool" not in result + + result = delegate_tool.run( + coworker="analyst", + task="Analyze data", + context="Sales data" + ) + assert "Error executing tool" in result + assert "analyst" not in result.lower() or "not found" in result.lower() + +def test_target_agents_none_allows_all(): + """Test that target_agents=None allows delegation to all agents.""" + researcher = Agent( + role="researcher", + goal="Research topics", + backstory="Expert researcher", + allow_delegation=True, + target_agents=None # Should allow all + ) + + writer = Agent( + role="writer", + goal="Write content", + backstory="Expert writer" + ) + + analyst = Agent( + role="analyst", + goal="Analyze data", + backstory="Expert analyst" + ) + + tools = researcher.get_delegation_tools([writer, analyst]) + delegate_tool = tools[0] + + result1 = delegate_tool.run(coworker="writer", task="Write", context="test") + result2 = delegate_tool.run(coworker="analyst", task="Analyze", context="test") + + assert "Error executing tool" not in result1 + assert "Error executing tool" not in result2 + +def test_target_agents_empty_list_blocks_all(): + """Test that target_agents=[] blocks delegation to all agents.""" + researcher = Agent( + role="researcher", + goal="Research topics", + backstory="Expert researcher", + allow_delegation=True, + target_agents=[] # Should block all + ) + + writer = Agent( + role="writer", + goal="Write content", + backstory="Expert writer" + ) + + tools = researcher.get_delegation_tools([writer]) + delegate_tool = tools[0] + + result = delegate_tool.run( + coworker="writer", + task="Write an article", + context="About AI" + ) + assert "Error executing tool" in result + +def test_target_agents_with_invalid_names(): + """Test behavior when target_agents contains invalid agent names.""" + researcher = Agent( + role="researcher", + goal="Research topics", + backstory="Expert researcher", + allow_delegation=True, + target_agents=["writer", "nonexistent_agent"] + ) + + writer = Agent( + role="writer", + goal="Write content", + backstory="Expert writer" + ) + + tools = researcher.get_delegation_tools([writer]) + delegate_tool = tools[0] + + result = delegate_tool.run( + coworker="writer", + task="Write an article", + context="About AI" + ) + assert "Error executing tool" not in result diff --git a/tests/tools/agent_tools/agent_tools_test.py b/tests/tools/agent_tools/agent_tools_test.py index 6cb5d26e7..2e8f3315f 100644 --- a/tests/tools/agent_tools/agent_tools_test.py +++ b/tests/tools/agent_tools/agent_tools_test.py @@ -131,3 +131,48 @@ def test_ask_question_to_wrong_agent(): result == "\nError executing tool. coworker mentioned not found, it must be one of the following options:\n- researcher\n" ) + +def test_target_agents_delegation_filtering(): + """Test that target_agents properly filters delegation targets.""" + researcher = Agent( + role="researcher", + goal="make the best research and analysis on content about AI and AI agents", + backstory="You're an expert researcher, specialized in technology", + allow_delegation=True, + target_agents=["writer"] # Can only delegate to writer + ) + + writer = Agent( + role="writer", + goal="Write engaging content", + backstory="You're an expert writer", + allow_delegation=False + ) + + analyst = Agent( + role="analyst", + goal="Analyze data trends", + backstory="You're an expert analyst", + allow_delegation=False + ) + + tools = AgentTools(agents=[writer]).tools() # Only writer available + delegate_tool = tools[0] + + result = delegate_tool.run( + coworker="writer", + task="Write about AI trends", + context="Latest developments" + ) + assert "Error executing tool" not in result + + tools_with_analyst = AgentTools(agents=[analyst]).tools() + delegate_tool_analyst = tools_with_analyst[0] + + result = delegate_tool_analyst.run( + coworker="analyst", + task="Analyze trends", + context="Data analysis" + ) + # This should work since we're passing analyst directly to AgentTools + assert "Error executing tool" not in result