A Coding Guide to Build a Hierarchical Supervisor Agent Framework with CrewAI and Google Gemini for Coordinated Multi-Agent Workflows

Building a Robust Supervisor Agent Framework with Google Gemini

This guide demonstrates how to architect and deploy a sophisticated Supervisor Agent Framework leveraging the Google Gemini model. We establish a team of specialized agents-researchers, analysts, writers, and reviewers-each assigned distinct responsibilities. These agents operate under the guidance of a central supervisor agent who orchestrates their collaboration, ensuring seamless workflow and consistent quality throughout the project lifecycle. By integrating structured task definitions, layered workflows, and embedded tools, this framework guarantees clarity in roles and oversight in execution.

Setting Up the Environment and Defining Task Priorities

To begin, install the necessary libraries and modules to power the CrewAI framework. We introduce a TaskPriority enumeration to categorize tasks by urgency and importance, enabling prioritized task management.

!pip install crewai crewai-tools langchain-google-genai python-dotenv

import os
from enum import Enum

class TaskPriority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

Configuring Task Parameters and Initializing the Supervisor Framework

We define a TaskConfig data class to encapsulate task details such as description, expected output, priority, maximum execution time, and whether human input is required. This standardization streamlines task handling across agents.

from dataclasses import dataclass

@dataclass
class TaskConfig:
    description: str
    expected_output: str
    priority: TaskPriority
    max_execution_time: int = 300  # seconds
    requires_human_input: bool = False

The SupervisorFramework class integrates the Google Gemini API and optional web search tools, setting up specialized agents and a supervisor to manage the project workflow.

Specialized Agents: Roles and Responsibilities

Each agent is tailored for a specific function:

• Research Agent: Gathers and verifies comprehensive information from credible sources.
• Analyst Agent: Interprets data to extract meaningful insights and trends.
• Writer Agent: Crafts clear, engaging, and structured content based on research and analysis.
• Reviewer Agent: Ensures deliverables meet quality standards through meticulous evaluation.
• Supervisor Agent: Oversees coordination, workflow management, and quality assurance across all agents.

Constructing a Hierarchical Task Workflow

The framework builds a sequential task pipeline where each stage depends on the previous one’s output, ensuring logical progression from research to final review. The supervisor agent monitors and harmonizes this process, resolving conflicts and maintaining standards.

def create_task_workflow(self, topic: str, task_configs: dict) -> list:
    tasks = []

    if 'research' in task_configs:
        config = task_configs['research']
        tasks.append(Task(
            description=f"{config.description} Topic: {topic}",
            expected_output=config.expected_output,
            agent=self.agents['researcher']
        ))

    if 'analysis' in task_configs:
        config = task_configs['analysis']
        tasks.append(Task(
            description=f"{config.description} Analyze research on: {topic}",
            expected_output=config.expected_output,
            agent=self.agents['analyst'],
            context=tasks
        ))

    if 'writing' in task_configs:
        config = task_configs['writing']
        tasks.append(Task(
            description=f"{config.description} Write content about: {topic}",
            expected_output=config.expected_output,
            agent=self.agents['writer'],
            context=tasks
        ))

    if 'review' in task_configs:
        config = task_configs['review']
        tasks.append(Task(
            description=f"{config.description} Review all outputs for: {topic}",
            expected_output=config.expected_output,
            agent=self.agents['reviewer'],
            context=tasks
        ))

    supervisor_task = Task(
        description=f"Supervise and coordinate all tasks for: {topic}. Ensure quality and resolve issues.",
        expected_output="Final project summary, quality report, and recommendations.",
        agent=self.supervisor,
        context=tasks
    )
    tasks.append(supervisor_task)

    return tasks

Sample Task Configurations for Effective Workflow Management

To illustrate, here are example task configurations that define clear objectives, deliverables, and priorities for each phase:

def create_sample_task_configs() -> dict:
    return {
        'research': TaskConfig(
            description="Perform in-depth research using authoritative sources.",
            expected_output="Comprehensive report with key data, trends, and citations (min 500 words).",
            priority=TaskPriority.HIGH
        ),
        'analysis': TaskConfig(
            description="Evaluate research data to uncover patterns and insights.",
            expected_output="Analytical summary highlighting opportunities and challenges (min 400 words).",
            priority=TaskPriority.HIGH
        ),
        'writing': TaskConfig(
            description="Develop a well-organized document synthesizing research and analysis.",
            expected_output="Professional report with actionable recommendations (min 800 words).",
            priority=TaskPriority.MEDIUM
        ),
        'review': TaskConfig(
            description="Conduct thorough quality checks on all deliverables.",
            expected_output="Quality assurance report with feedback and approval status.",
            priority=TaskPriority.CRITICAL
        )
    }

Demonstration: Running the Supervisor Framework

The following function demonstrates initializing the framework, setting up agents, defining a project topic, and executing the workflow. It outputs progress updates, final results, and resource usage metrics.

def demo_supervisor_framework():
    print("🚀 Starting CrewAI Supervisor Framework Demo")
    print("=" * 50)

    framework = SupervisorFramework(
        gemini_api_key="Your_Google_Gemini_API_Key",
        serper_api_key=None
    )

    topic = "Emerging Trends in Renewable Energy 2024"
    task_configs = create_sample_task_configs()

    print(f"📊 Project Topic: {topic}")
    print(f"🗂️ Task Types: {list(task_configs.keys())}")

    results = framework.execute_project(topic, task_configs)

    print("n" + "=" * 50)
    print("📈 EXECUTION SUMMARY")
    print("=" * 50)

    if results['status'] == 'success':
        print(f"✅ Status: {results['status'].upper()}")
        print(f"📝 Tasks Completed: {results['tasks_completed']}")
        print(f"🤖 Agents Engaged: {results['agents_involved']}")
        print(f"📄 Sample Output Preview: {str(results['result'])[:200]}...")
    else:
        print(f"❌ Status: {results['status'].upper()}")
        print(f"🚫 Error: {results['error']}")

    metrics = framework.get_crew_usage_metrics()
    print(f"n💰 Resource Usage: {metrics}")

if __name__ == "__main__":
    demo_supervisor_framework()

Conclusion: Streamlining Complex Projects with Coordinated AI Agents

This Supervisor Agent Framework exemplifies how to efficiently manage multifaceted projects by harnessing a team of specialized AI agents working in harmony. The structured workflow-from research through review-ensures each phase is executed with precision, while the supervisor agent maintains oversight and quality control. This approach transforms abstract objectives into tangible, high-quality outputs, making it ideal for real-world applications requiring collaboration, accuracy, and scalability.