andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-09 08:08:32 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Fix delegate coworker bug in hierarchical mode and add test

Browse Source 
- Update _prepare_tools and _update_manager_tools to handle async execution in hierarchical mode
- Add test cases to verify delegation tool behavior
- Ensure proper tool updates during async task execution

Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
Devin AI
2024-12-30 19:26:01 +00:00
parent 73f328860b
commit 0a137ca9dd
2 changed files with 159 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/crewai/crew.py
View File

@@ -802,11 +802,21 @@ class Crew(BaseModel):
if agent.allow_delegation:
if self.process == Process.hierarchical:
if self.manager_agent:
tools = self._update_manager_tools(task, tools)
# For hierarchical process, handle both manager and regular agent tools
if agent == self.manager_agent:
# Manager can delegate to all regular agents
tools = self._inject_delegation_tools(tools, agent, [a for a in self.agents if a != agent])
else:
# For async tasks, maintain original delegation behavior
if task and task.async_execution:
tools = self._add_delegation_tools(task, tools)
else:
# Regular agents can delegate to manager and other agents
delegation_agents = [self.manager_agent] + [a for a in self.agents if a != agent]
tools = self._inject_delegation_tools(tools, agent, delegation_agents)
else:
raise ValueError("Manager agent is required for hierarchical process.")
elif agent and agent.allow_delegation:
else:
tools = self._add_delegation_tools(task, tools)
# Add code execution tools if agent allows code execution
@@ -852,8 +862,17 @@ class Crew(BaseModel):
return self._merge_tools(tools, code_tools)
def _add_delegation_tools(self, task: Task, tools: List[Tool]):
agents_for_delegation = [agent for agent in self.agents if agent != task.agent]
if len(self.agents) > 1 and len(agents_for_delegation) > 0 and task.agent:
agents_for_delegation = []
if self.process == Process.hierarchical and self.manager_agent:
# In hierarchical mode, allow delegation to manager if the task agent isn't the manager
if task.agent != self.manager_agent:
agents_for_delegation = [self.manager_agent]
# Also include regular agents for delegation
agents_for_delegation.extend([agent for agent in self.agents if agent != task.agent])
else:
agents_for_delegation = [agent for agent in self.agents if agent != task.agent]
if len(agents_for_delegation) > 0 and task.agent:
if not tools:
tools = []
tools = self._inject_delegation_tools(tools, task.agent, agents_for_delegation)
@@ -868,7 +887,21 @@ class Crew(BaseModel):
def _update_manager_tools(self, task: Task, tools: List[Tool]):
if self.manager_agent:
if task.agent:
tools = self._inject_delegation_tools(tools, task.agent, [task.agent])
# For async tasks, maintain original delegation behavior
if task.async_execution:
tools = self._add_delegation_tools(task, tools)
else:
# In hierarchical mode, allow bidirectional delegation
if self.process == Process.hierarchical:
# For non-manager agents, allow delegation to manager and other agents
if task.agent != self.manager_agent:
delegation_agents = [self.manager_agent] + self.agents
tools = self._inject_delegation_tools(tools, task.agent, [a for a in delegation_agents if a != task.agent])
else:
# For manager, allow delegation to all agents
tools = self._inject_delegation_tools(tools, task.agent, [a for a in self.agents if a != task.agent])
else:
tools = self._inject_delegation_tools(tools, task.agent, self.agents)
else:
tools = self._inject_delegation_tools(tools, self.manager_agent, self.agents)
return tools

120
tests/test_hierarchical_delegation.py Normal file
View File

@@ -0,0 +1,120 @@
import pytest
from unittest.mock import MagicMock
from crewai import Agent, Task, Crew, Process
from langchain_core.language_models.base import BaseLanguageModel
def test_hierarchical_delegation_tool_availability():
"""Test that all agents are available for delegation in hierarchical mode."""
# Mock LLM to avoid actual API calls
mock_llm = MagicMock(spec=BaseLanguageModel)
# Create agents
manager = Agent(
role="Manager",
goal="Manage the team",
backstory="I am a manager",
allow_delegation=True,
llm=mock_llm
)
kb_agent = Agent(
role="kb_retriever_agent",
goal="Retrieve knowledge",
backstory="I am a knowledge retrieval specialist",
allow_delegation=True,
llm=mock_llm
)
worker = Agent(
role="Worker",
goal="Do the work",
backstory="I am a worker",
allow_delegation=True,
llm=mock_llm
)
# Create a task assigned to the manager
task = Task(
description="Complex task requiring delegation",
expected_output="Task completion status",
agent=manager
)
# Create the crew with hierarchical process
crew = Crew(
agents=[kb_agent, worker], # Manager should not be in agents list when using hierarchical process
tasks=[task],
process=Process.hierarchical,
manager_agent=manager # Explicitly set the manager agent for hierarchical process
)
# Get the manager's tools
tools_for_task = task.tools or manager.tools or []
manager_tools = crew._prepare_tools(manager, task, tools_for_task)
# Find delegation tools
delegation_tools = [tool for tool in manager_tools if tool.name == "Delegate work to coworker"]
assert len(delegation_tools) > 0, "Delegation tool should be present"
# Get the delegation tool description
delegate_tool = delegation_tools[0]
tool_description = str(delegate_tool.description)
# Verify all agents are available for delegation
assert "kb_retriever_agent" in tool_description, "kb_retriever_agent should be available for delegation"
assert "Worker" in tool_description, "Worker should be available for delegation"
def test_hierarchical_delegation_tool_updates():
"""Test that delegation tools are properly updated when task agent changes."""
mock_llm = MagicMock(spec=BaseLanguageModel)
manager = Agent(
role="Manager",
goal="Manage the team",
backstory="I am a manager",
allow_delegation=True,
llm=mock_llm
)
kb_agent = Agent(
role="kb_retriever_agent",
goal="Retrieve knowledge",
backstory="I am a knowledge retrieval specialist",
allow_delegation=True,
llm=mock_llm
)
# Create tasks for different agents
manager_task = Task(
description="Manager task",
expected_output="Manager task completion status",
agent=manager
)
kb_task = Task(
description="KB task",
expected_output="KB task completion status",
agent=kb_agent
)
# Create crew
crew = Crew(
agents=[kb_agent], # Manager should not be in agents list when using hierarchical process
tasks=[manager_task, kb_task],
process=Process.hierarchical,
manager_agent=manager # Explicitly set the manager agent for hierarchical process
)
# Test manager's tools
manager_tools_for_task = manager_task.tools or manager.tools or []
manager_tools = crew._prepare_tools(manager, manager_task, manager_tools_for_task)
manager_delegation = [t for t in manager_tools if t.name == "Delegate work to coworker"]
assert len(manager_delegation) > 0, "Manager should have delegation tool"
assert "kb_retriever_agent" in str(manager_delegation[0].description), "Manager should see kb_retriever_agent"
# Test kb_agent's tools
kb_tools_for_task = kb_task.tools or kb_agent.tools or []
kb_tools = crew._prepare_tools(kb_agent, kb_task, kb_tools_for_task)
kb_delegation = [t for t in kb_tools if t.name == "Delegate work to coworker"]
assert len(kb_delegation) > 0, "KB agent should have delegation tool"
assert "Manager" in str(kb_delegation[0].description), "KB agent should see manager"