QUESTION: "Force" complex hierarchy within the crew

[emilmirzayev]

Hey there. Newly discovered this project and I am having fun with it. I looked at process=Process.hierarchical in Crew settings. I have a question about it.

Consider a case where there are four agents: Agent 1 - 5, and four tasks: Task 1 - 5.

There must be a specific hierarchy in the crew:

Agent 1
├── Agent 2
│ ├── Agent 3
│ └── Agent 4
└── Agent 5

When I want to build a crew like this, I can write my code in the following way:

class MYagents:

    def agent1(self):

        return Agent(
            role="Sample role 1",
            goal="Sample goal 1",
            backstory=dedent("""
                Sample backstory for Agent 1.
            """),
            verbose=False,
            llm=sample_llm,
            allow_delegation= True,
        )

    def agent2(self):

        return Agent(
            role="Sample role 2",
            goal="Sample goal 2",
            backstory=dedent("""
                Sample backstory for Agent 2
            """),
            verbose=False,
            llm=sample_llm,
            allow_delegation= True,
        )

    def agent3(self):
        return Agent(
            role="Sample role 3",
            goal="Sample goal 3",
            backstory=dedent("""
                Sample backstory for Agent 3
            """),
            verbose=False,
            llm=sample_llm,
        )

    def agent4(self):
        return Agent(
            role="Sample role 4",
            goal="Sample goal 4",
            backstory=dedent("""
                Sample backstory for Agent 4
            """),
            verbose=False,
            llm=sample_llm,
        )

    def agent5(self):
        return Agent(
            role="Sample role 5",
            goal="Sample goal 5",
            backstory=dedent("""
                Sample backstory for Agent 5
            """),
            verbose=False,
            llm=sample_llm,
        )
# Defining a class for tasks with generic placeholder methods
class MYtasks:

    def task1(self, agent, context):
        return Task(
            description=dedent("""
                Sample task 1 description for Agent 1.
            """),
            expected_output=dedent("""
                Sample expected output for Task 1.
            """),
            agent=agent,
            context = context
        )

    def task2(self, agent, context):
        return Task(
            description=dedent("""
                Sample task 2 description for Agent 2.
            """),
            expected_output=dedent("""
                Sample expected output for Task 2.
            """),
            agent=agent,
            context = context
        )

    def task3(self, agent):
        return Task(
            description=dedent("""
                Sample task 3 description for Agent 3.
            """),
            expected_output=dedent("""
                Sample expected output for Task 3.
            """),
            agent=agent,
        )

    def task4(self, agent):
        return Task(
            description=dedent("""
                Sample task 4 description for Agent 4.
            """),
            expected_output=dedent("""
                Sample expected output for Task 4.
            """),
            agent=agent,
        )

    def task5(self, agent):
        return Task(
            description=dedent("""
                Sample task 5 description for Agent 5.
            """),
            expected_output=dedent("""
                Sample expected output for Task 5.
            """),
            agent=agent,
        )

agents = MYagents()
tasks = MYtasks()

# Creating agents
agent_1 = agents.agent1()
agent_2 = agents.agent2()
agent_3 = agents.agent3()
agent_4 = agents.agent4()
agent_5 = agents.agent5()

# Assigning tasks to agents

task_3 = tasks.task3(agent_3)
task_4 = tasks.task4(agent_4)
task_5 = tasks.task5(agent_5)
task_2 = tasks.task2(agent_2, context = [task_3, task_4])
task_1 = tasks.task1(agent_1, context = [task_2, task_5])

crew = Crew(
    agents=[
        agent1,  # Assuming Agent 1 is the manager in this context
        agent2,
        agent3,
        agent4,
        agent_5
    ],
    tasks=[
        task1, 
        task2,  
        task3,  
        task4  
        task5   
    ],
    verbose=False,
    process=Process.hierarchical,
    manager_llm=sample_llm
)

However, this brings a question. How can I make agent1 to be the manager. Because it has its own goals and instructions which manager_llm will not have. How can one achieve that? The labor division is made by specific agent. How do I need to change my definitions to achieve such outcome?

thank you for anyone who clarifies this for me.

[francisjervis]

Looking at a similar problem and I think the best way to achieve this would be to allow custom prompts to be passed to the "manager" agent in a hierarchical process.

[noggynoggy]

@francisjervis I am working on an implementation for that. Basically, you pass in a manager_agent into your crew, and it will be used in .venv/lib/python3.11/site-packages/crewai/crew.py to create the manager. This allows an explicit manager, but not the explicit structure @emilmirzayev asked for out of the box. But you could go further. To explain, read this (you need to modify the lib for this to work):

def get_manager():
    return = Agent(
        role="Crew Manager",
        goal="Manage the team to complete the task in the best way possible.",
        backstory="""You are a seasoned [...].""",
        allow_delegation=True, # has to be true
        tools=AgentTools(agents=get_agents(expertise=expertise)).tools(),
        llm=llm,
        max_iter=10,
        verbose=True,
    )

def get_crew():
    return Crew(
        agents = get_agents(),
        tasks = get_tasks(),
        manager_llm = llm,
        manager_agent = get_manager(), 
        process = Process.hierarchical,
    )

As you can see, the tools of the manager agent takes in a list of tools. By default, these are Delegate work to co-worker, and Ask question to co-worker. The function AgentTools takes in a list of agents, which are then accessible for the manager. So @emilmirzayev could then just make Agent1 to manager, and pass in only Agent2 and Agent5, so the manager can only delegate to them. This would get you at least half way there.

This works right now on my machine™ and tomorrow I will look into opening a PR to get the changes into the project. But I'll have to look into contribution guidelines first.

[francisjervis]

I did it another way - adding manager_goal etc parameters to the Crew class which (if present) are used instead of the strings file defaults. This may be more straightforward for most implementations, though being able to pass arbitrary Langchain agents would be useful...

[emilmirzayev]

@francisjervis I would love to have some more information about your solution. could you share some sample code?

[emilmirzayev]

@noggynoggy , interesting approach. How do you plan to add manager_agent to Crew? Overwrite _run_hierarchical_process 'sllm ?

[noggynoggy]

@noggynoggy , interesting approach. How do you plan to add manager_agent to Crew? Overwrite _run_hierarchical_process 'sllm ?

474

[joaomdmoura]

474 Is merged, I like this idea! We will add some tests next before we cut a version with it, but should come out this week

[francisjervis]

@francisjervis I would love to have some more information about your solution. could you share some sample code? Add three parameters to the Crew class in crew.py :

class Crew(BaseModel):
"""
Represents a group of agents, defining how they should collaborate and the tasks they should perform.
This class allows customization of the manager's role, goal, and backstory by optionally accepting these values upon initialization.

Attributes:
    tasks: List of tasks assigned to the crew.
    agents: List of agents part of this crew.
    manager_llm: The language model that will run the manager agent.
    manager_role: Optional custom role for the manager.
    manager_goal: Optional custom goal for the manager.
    manager_backstory: Optional custom backstory for the manager.
    function_calling_llm: The language model that will run the tool calling for all the agents.
    process: The process flow that the crew will follow (e.g., sequential).
    verbose: Indicates the verbosity level for logging during execution.
    config: Configuration settings for the crew.
    max_rpm: Maximum number of requests per minute for the crew execution to be respected.
    id: A unique identifier for the crew instance.
    full_output: Whether the crew should return the full output with all tasks outputs or just the final output.
    step_callback: Callback to be executed after each step for every agents execution.
    share_crew: Whether you want to share the complete crew information and execution with crewAI to make the library better, and allow us to train models.
    language: Language used for the crew, defaults to English.
"""

__hash__ = object.__hash__  # type: ignore
_execution_span: Any = PrivateAttr()
_rpm_controller: RPMController = PrivateAttr()
_logger: Logger = PrivateAttr()
_cache_handler: InstanceOf[CacheHandler] = PrivateAttr(default=CacheHandler())
model_config = ConfigDict(arbitrary_types_allowed=True)
tasks: List[Task] = Field(default_factory=list)
agents: List[Agent] = Field(default_factory=list)
process: Process = Field(default=Process.sequential)
verbose: Union[int, bool] = Field(default=0)
usage_metrics: Optional[dict] = Field(
    default=None,
    description="Metrics for the LLM usage during all tasks execution.",
)
full_output: Optional[bool] = Field(
    default=False,
    description="Whether the crew should return the full output with all tasks outputs or just the final output.",
)
manager_llm: Optional[Any] = Field(
    default=None,
    description="Language model that will run the manager agent.",
)
manager_role: Optional[str] = Field(
    default=None,
    description="Custom role for the manager agent."
)
manager_goal: Optional[str] = Field(
    default=None,
    description="Custom goal for the manager agent."
)
manager_backstory: Optional[str] = Field(
    default=None,
    description="Custom backstory for the manager agent."
)

etc

Then further down change the start of `_run_hierarchical_process`:

def _run_hierarchical_process(self) -> str:
    """Creates and assigns a manager agent to make sure the crew completes the tasks."""

    i18n = I18N(language=self.language)

    role = self.manager_role or i18n.retrieve("hierarchical_manager_agent", "role")
    goal = self.manager_goal or i18n.retrieve("hierarchical_manager_agent", "goal")
    backstory = self.manager_backstory or i18n.retrieve("hierarchical_manager_agent", "backstory")

[github-actions[bot]]

This issue is stale because it has been open for 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.

[github-actions[bot]]

This issue was closed because it has been stalled for 5 days with no activity.