From 8fd28671551ef62cb8ef33ae9665cbe97be58793 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 17 Jun 2025 16:33:52 +0000 Subject: [PATCH] feat: implement hierarchical agent delegation with allowed_agents parameter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add allowed_agents field to BaseAgent class with validation - Modify AgentTools to filter delegation targets based on allowed_agents - Update Agent.get_delegation_tools to pass delegating agent context - Support both role strings and agent instances in allowed_agents - Implement case-insensitive role matching for flexibility - Add comprehensive test coverage for all scenarios - Maintain backward compatibility (None = allow all agents) - Handle edge cases (empty list = no delegation allowed) Addresses issue #2068 for controlled hierarchical delegation Co-Authored-By: João --- src/crewai/agent.py | 2 +- src/crewai/agents/agent_builder/base_agent.py | 22 +- src/crewai/tools/agent_tools/agent_tools.py | 42 ++- tests/test_allowed_agents.py | 263 ++++++++++++++++++ 4 files changed, 322 insertions(+), 7 deletions(-) create mode 100644 tests/test_allowed_agents.py diff --git a/src/crewai/agent.py b/src/crewai/agent.py index c8e34b2e6..edf8a1049 100644 --- a/src/crewai/agent.py +++ b/src/crewai/agent.py @@ -565,7 +565,7 @@ class Agent(BaseAgent): def get_delegation_tools(self, agents: List[BaseAgent]): agent_tools = AgentTools(agents=agents) - tools = agent_tools.tools() + tools = agent_tools.tools(delegating_agent=self) return tools def get_multimodal_tools(self) -> Sequence[BaseTool]: diff --git a/src/crewai/agents/agent_builder/base_agent.py b/src/crewai/agents/agent_builder/base_agent.py index ba2596f63..32c3f9010 100644 --- a/src/crewai/agents/agent_builder/base_agent.py +++ b/src/crewai/agents/agent_builder/base_agent.py @@ -2,7 +2,7 @@ import uuid from abc import ABC, abstractmethod from copy import copy as shallow_copy from hashlib import md5 -from typing import Any, Callable, Dict, List, Optional, TypeVar +from typing import Any, Callable, Dict, List, Optional, TypeVar, Union from pydantic import ( UUID4, @@ -108,6 +108,10 @@ class BaseAgent(ABC, BaseModel): default=False, description="Enable agent to delegate and ask questions among each other.", ) + allowed_agents: Optional[List[Union[str, "BaseAgent"]]] = Field( + default=None, + description="List of agent roles or agent instances that this agent can delegate to. If None, can delegate to all agents when allow_delegation=True.", + ) tools: Optional[List[BaseTool]] = Field( default_factory=list, description="Tools at agents' disposal" ) @@ -195,6 +199,22 @@ class BaseAgent(ABC, BaseModel): ) return processed_tools + @field_validator("allowed_agents") + @classmethod + def validate_allowed_agents(cls, allowed_agents: Optional[List[Union[str, "BaseAgent"]]]) -> Optional[List[Union[str, "BaseAgent"]]]: + """Validate the allowed_agents list.""" + if allowed_agents is None: + return None + + if not isinstance(allowed_agents, list): + raise ValueError("allowed_agents must be a list of agent roles (strings) or agent instances") + + for agent in allowed_agents: + if not isinstance(agent, (str, BaseAgent)): + raise ValueError("Each item in allowed_agents must be either a string (agent role) or a BaseAgent instance") + + return allowed_agents + @model_validator(mode="after") def validate_and_set_attributes(self): # Validate required fields diff --git a/src/crewai/tools/agent_tools/agent_tools.py b/src/crewai/tools/agent_tools/agent_tools.py index 77d3c2d89..82818f45b 100644 --- a/src/crewai/tools/agent_tools/agent_tools.py +++ b/src/crewai/tools/agent_tools/agent_tools.py @@ -1,3 +1,5 @@ +from typing import Optional + from crewai.agents.agent_builder.base_agent import BaseAgent from crewai.tools.base_tool import BaseTool from crewai.utilities import I18N @@ -13,20 +15,50 @@ class AgentTools: self.agents = agents self.i18n = i18n - def tools(self) -> list[BaseTool]: - """Get all available agent tools""" - coworkers = ", ".join([f"{agent.role}" for agent in self.agents]) + def tools(self, delegating_agent: Optional[BaseAgent] = None) -> list[BaseTool]: + """Get all available agent tools, filtered by delegating agent's allowed_agents if specified""" + available_agents = self._filter_allowed_agents(delegating_agent) + + if not available_agents: + return [] + + coworkers = ", ".join([f"{agent.role}" for agent in available_agents]) delegate_tool = DelegateWorkTool( - agents=self.agents, + agents=available_agents, i18n=self.i18n, description=self.i18n.tools("delegate_work").format(coworkers=coworkers), # type: ignore ) ask_tool = AskQuestionTool( - agents=self.agents, + agents=available_agents, i18n=self.i18n, description=self.i18n.tools("ask_question").format(coworkers=coworkers), # type: ignore ) return [delegate_tool, ask_tool] + + def _filter_allowed_agents(self, delegating_agent: Optional[BaseAgent]) -> list[BaseAgent]: + """Filter agents based on the delegating agent's allowed_agents list""" + if delegating_agent is None: + return self.agents + + if not hasattr(delegating_agent, 'allowed_agents') or delegating_agent.allowed_agents is None: + return self.agents + + if not delegating_agent.allowed_agents: + return [] + + filtered_agents = [] + for agent in self.agents: + for allowed in delegating_agent.allowed_agents: + if isinstance(allowed, str): + if agent.role.lower() == allowed.lower(): + filtered_agents.append(agent) + break + elif isinstance(allowed, BaseAgent): + if agent is allowed: + filtered_agents.append(agent) + break + + return filtered_agents diff --git a/tests/test_allowed_agents.py b/tests/test_allowed_agents.py new file mode 100644 index 000000000..e4a1db4e4 --- /dev/null +++ b/tests/test_allowed_agents.py @@ -0,0 +1,263 @@ +"""Test allowed_agents functionality for hierarchical delegation.""" + +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 + + +@pytest.fixture +def agents(): + """Create test agents for delegation testing.""" + manager = Agent( + role="Manager", + goal="Manage the team", + backstory="You are a team manager", + allow_delegation=True, + ) + + researcher = Agent( + role="Researcher", + goal="Research topics", + backstory="You are a researcher", + allow_delegation=False, + ) + + writer = Agent( + role="Writer", + goal="Write content", + backstory="You are a writer", + allow_delegation=False, + ) + + analyst = Agent( + role="Analyst", + goal="Analyze data", + backstory="You are an analyst", + allow_delegation=False, + ) + + return manager, researcher, writer, analyst + + +def test_allowed_agents_with_role_strings(agents): + """Test allowed_agents with role strings.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = ["Researcher", "Writer"] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + assert len(tools) == 2 + delegate_tool = tools[0] + + assert len(delegate_tool.agents) == 2 + agent_roles = [agent.role for agent in delegate_tool.agents] + assert "Researcher" in agent_roles + assert "Writer" in agent_roles + assert "Analyst" not in agent_roles + + +def test_allowed_agents_with_agent_instances(agents): + """Test allowed_agents with agent instances.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = [researcher, analyst] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + assert len(tools) == 2 + delegate_tool = tools[0] + + assert len(delegate_tool.agents) == 2 + assert researcher in delegate_tool.agents + assert analyst in delegate_tool.agents + assert writer not in delegate_tool.agents + + +def test_allowed_agents_mixed_types(agents): + """Test allowed_agents with mixed role strings and agent instances.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = ["Researcher", writer] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 2 + assert researcher in delegate_tool.agents + assert writer in delegate_tool.agents + assert analyst not in delegate_tool.agents + + +def test_allowed_agents_empty_list(agents): + """Test allowed_agents with empty list (no delegation allowed).""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = [] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + assert len(tools) == 0 + + +def test_allowed_agents_none(agents): + """Test allowed_agents with None (delegate to all agents).""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = None + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 3 + assert researcher in delegate_tool.agents + assert writer in delegate_tool.agents + assert analyst in delegate_tool.agents + + +def test_allowed_agents_case_insensitive_matching(agents): + """Test that role matching is case-insensitive.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = ["researcher", "WRITER"] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 2 + assert researcher in delegate_tool.agents + assert writer in delegate_tool.agents + + +def test_allowed_agents_validation(): + """Test validation of allowed_agents field.""" + agent = Agent( + role="Test", + goal="Test", + backstory="Test", + allowed_agents=["Role1", "Role2"] + ) + assert agent.allowed_agents == ["Role1", "Role2"] + + agent = Agent( + role="Test", + goal="Test", + backstory="Test", + allowed_agents=None + ) + assert agent.allowed_agents is None + + with pytest.raises(ValueError, match="allowed_agents must be a list"): + Agent( + role="Test", + goal="Test", + backstory="Test", + allowed_agents="invalid" + ) + + with pytest.raises(ValueError, match="must be either a string"): + Agent( + role="Test", + goal="Test", + backstory="Test", + allowed_agents=[123] + ) + + +def test_crew_integration_with_allowed_agents(agents): + """Test integration with Crew class.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = ["Researcher"] + + task = Task( + description="Research AI trends", + expected_output="Research report", + agent=manager + ) + + crew = Crew( + agents=[manager, researcher, writer, analyst], + tasks=[task] + ) + + tools = crew._prepare_tools(manager, task, []) + + delegation_tools = [tool for tool in tools if "Delegate work" in tool.name or "Ask question" in tool.name] + + if delegation_tools: + delegate_tool = next(tool for tool in delegation_tools if "Delegate work" in tool.name) + assert len(delegate_tool.agents) == 1 + assert delegate_tool.agents[0].role == "Researcher" + + +def test_backward_compatibility_no_allowed_agents(agents): + """Test that agents without allowed_agents work as before.""" + manager, researcher, writer, analyst = agents + + assert manager.allowed_agents is None + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + assert len(tools) == 2 + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 3 + assert researcher in delegate_tool.agents + assert writer in delegate_tool.agents + assert analyst in delegate_tool.agents + + +def test_no_delegating_agent_parameter(agents): + """Test AgentTools.tools() without delegating_agent parameter.""" + manager, researcher, writer, analyst = agents + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools() # No delegating_agent parameter + + assert len(tools) == 2 + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 3 + + +def test_allowed_agents_with_nonexistent_role(agents): + """Test allowed_agents with role that doesn't exist in available agents.""" + manager, researcher, writer, analyst = agents + + manager.allowed_agents = ["Researcher", "NonExistentRole"] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 1 + assert researcher in delegate_tool.agents + + +def test_allowed_agents_with_nonexistent_instance(agents): + """Test allowed_agents with agent instance that doesn't exist in available agents.""" + manager, researcher, writer, analyst = agents + + other_agent = Agent( + role="Other", + goal="Other goal", + backstory="Other backstory" + ) + + manager.allowed_agents = [researcher, other_agent] + + agent_tools = AgentTools(agents=[researcher, writer, analyst]) + tools = agent_tools.tools(delegating_agent=manager) + + delegate_tool = tools[0] + assert len(delegate_tool.agents) == 1 + assert researcher in delegate_tool.agents + assert other_agent not in delegate_tool.agents