Hierarchical Multi-Agent System 🤖

A modular implementation of a hierarchical multi-agent system using LangGraph and LangChain, featuring CEO-led coordination of specialized research and writing teams for comprehensive task analysis and report generation.

📋 Table of Contents

• Overview
• Architecture
• Features
• Project Structure
• Getting Started
• Usage
• Agent Hierarchy
• Workflow Stages
• Configuration
• Development Guide
• Testing
• Contributing

🎯 Overview

This project implements a sophisticated hierarchical multi-agent system where a CEO agent coordinates specialized teams to complete complex tasks. The system demonstrates:

• Hierarchical Decision Making: CEO orchestrates high-level workflow and delegates to team leaders
• Team-Based Specialization: Dedicated research and writing teams with specialized agents
• State Management: Centralized state tracking using LangGraph's MessagesState
• Dynamic Routing: Intelligent workflow routing based on task completion status
• Modular Architecture: Clean separation of concerns with extensible design

Use Cases

• Comprehensive market research and analysis
• Technical documentation generation
• Strategic business reports
• Industry trend analysis
• Multi-perspective content creation

🏗️ Architecture

┌─────────────────────────────────────────────────────────┐
│                      CEO AGENT                          │
│         (Orchestration & Decision Making)               │
└────────────────┬─────────────────┬──────────────────────┘
                 │                 │
      ┌──────────▼─────────┐    ┌─▼──────────────────┐
      │  Research Team     │    │  Writing Team      │
      │  Leader            │    │  Leader            │
      └──┬─────────────┬───┘    └──┬────────────┬────┘
         │             │            │            │
    ┌────▼───┐   ┌────▼────┐  ┌───▼─────┐ ┌────▼────────┐
    │ Data   │   │ Market  │  │ Summary │ │ Technical   │
    │Research│   │Research │  │ Writer  │ │ Writer      │
    └────────┘   └─────────┘  └─────────┘ └─────────────┘
                         │
                    ┌────▼─────────────┐
                    │  Consolidation   │
                    │     Agent        │
                    └──────────────────┘

Workflow Stages

1. Planning - CEO receives task and initiates workflow
2. Research Phase - Research team gathers data and market insights
3. Writing Phase - Writing team creates summary and technical content
4. Consolidation - Final report generation combining all outputs

✨ Features

Core Capabilities

• Hierarchical Agent Architecture: Three-tier hierarchy (CEO → Team Leaders → Specialists)
• Modular Design: Each agent is independently implemented and testable
• State Management: Robust state tracking with HierarchicalState TypedDict
• Dynamic Routing: Conditional edge routing based on completion status
• Team Coordination: Parallel team operations with synchronized reporting
• LLM Integration: Configurable LLM support (Groq, OpenAI, etc.)
• Web UI: Streamlit interface for interactive task execution
• Extensible Framework: Easy to add new agents, teams, or workflow stages

Agent Specializations

Research Team

• Data Researcher: Gathers factual information, statistics, and quantitative data
• Market Researcher: Analyzes market trends, competition, and opportunities

Writing Team

• Summary Writer: Creates executive summaries and high-level overviews
• Technical Writer: Develops detailed technical documentation and analysis

📁 Project Structure

hierarchical_agent_system/
├── main.py                     # Entry point with CLI interface
├── requirements.txt            # Python dependencies
├── README.md                   # Project documentation
├── .gitignore                  # Git ignore rules
├── raw_agent_book.ipynb       # Development notebook
│
├── src/                        # Source code directory
│   ├── __init__.py
│   │
│   ├── core/                   # Core system components
│   │   ├── __init__.py
│   │   ├── state.py           # HierarchicalState definition
│   │   ├── router.py          # Conditional routing logic
│   │   ├── workflow.py        # LangGraph workflow builder
│   │   └── llm_config.py      # LLM configuration and chains
│   │
│   ├── agents/                 # Individual agent implementations
│   │   ├── __init__.py
│   │   ├── ceo.py             # CEO orchestration agent
│   │   ├── data_researcher.py  # Data research specialist
│   │   ├── market_researcher.py # Market analysis specialist
│   │   ├── summary_writer.py   # Executive summary specialist
│   │   ├── technical_writer.py # Technical documentation specialist
│   │   └── consolidation.py    # Final report consolidation
│   │
│   ├── teams/                  # Team leader implementations
│   │   ├── __init__.py
│   │   ├── research_team.py    # Research team coordinator
│   │   └── writing_team.py     # Writing team coordinator
│   │
│   └── ui/                     # User interface components
│       ├── __init__.py
│       ├── streamlit_app.py    # Streamlit web interface
│       └── components.py       # Reusable UI components
│
├── config/                     # Configuration files
│   ├── settings.py            # Application settings
│   └── .env.example           # Environment template
│
└── tests/                      # Test files
    ├── __init__.py
    ├── test_ceo.py            # CEO agent tests
    ├── test_research_team.py  # Research team tests
    ├── test_writing_team.py   # Writing team tests
    └── test_workflow.py       # Integration tests

🚀 Getting Started

Prerequisites

• Python 3.8 or higher
• pip package manager
• API key for LLM provider (Groq, OpenAI, etc.)

Installation

1. Clone the repository:

   git clone https://github.com/Arshp-svg/hierarchical-agent-system.git
   cd hierarchical-agent-system

2. Install dependencies:

   pip install -r requirements.txt

3. Set up environment variables:

   cp config/.env.example config/.env

   Edit config/.env and add your API keys:

   GROQ_API_KEY=your_groq_api_key_here
   # or
   OPENAI_API_KEY=your_openai_api_key_here

Quick Start

Run the default example:

python main.py

This will execute the system with the default task: "Analyze the impact of AI on healthcare industry"

💻 Usage

Command Line Interface

Run with default task:

python main.py

Run with custom task (modify main.py):

from main import run_custom_task

response = run_custom_task("Analyze the electric vehicle market in 2024")
print(response["final_report"])

Streamlit Web Interface

Launch the interactive web UI:

streamlit run src/ui/streamlit_app.py

Features:

• Task input interface
• Real-time workflow visualization
• Agent message tracking
• Final report display
• Export functionality

Programmatic Usage

from langchain_core.messages import HumanMessage
from src.core.workflow import create_hierarchical_workflow

# Create workflow
workflow = create_hierarchical_workflow()

# Define task
task = "Your custom analysis task here"

# Execute workflow
response = workflow.invoke(
    {"messages": [HumanMessage(content=task)]},
    config={"recursion_limit": 50}
)

# Access results
final_report = response.get("final_report", "")
research_data = response.get("research_team_report", "")
writing_data = response.get("writing_team_report", "")

🎭 Agent Hierarchy

CEO Agent (ceo.py)

Responsibilities:

• Receives initial task
• Monitors team completion status
• Routes workflow between research and writing phases
• Initiates consolidation when all teams complete

Decision Logic:

if research_completed and writing_completed:
    → consolidation
elif not research_completed:
    → research_team_leader
elif research_completed and not writing_completed:
    → writing_team_leader

Research Team Leader (research_team.py)

Responsibilities:

• Coordinates data and market researchers
• Tracks research task completion
• Compiles research reports
• Reports back to CEO

Workflow:

1. Assigns data research task
2. Assigns market research task
3. Compiles combined report
4. Updates team_completed["research_team"] = True

Writing Team Leader (writing_team.py)

Responsibilities:

• Coordinates summary and technical writers
• Manages writing task distribution
• Compiles writing outputs
• Reports back to CEO

Workflow:

1. Assigns summary writing task
2. Assigns technical writing task
3. Compiles combined document
4. Updates team_completed["writing_team"] = True

Specialist Agents

Data Researcher

• Gathers statistics and quantitative data
• Provides historical trends
• Delivers factual background information

Market Researcher

• Analyzes market size and growth
• Performs competitive analysis
• Identifies opportunities and challenges

Summary Writer

• Creates executive overviews
• Highlights key findings
• Provides strategic recommendations

Technical Writer

• Develops detailed technical content
• Provides implementation specifics
• Creates comprehensive documentation

Consolidation Agent (consolidation.py)

Responsibilities:

• Combines all team outputs
• Generates final comprehensive report
• Sets task_completed = True
• Triggers workflow termination

🔄 Workflow Stages

1. Planning Stage

User Input → CEO Agent → Task Analysis → Team Assignment

2. Research Phase

CEO → Research Team Leader → {Data Researcher, Market Researcher}
                           → Compile Research Report → CEO

3. Writing Phase

CEO → Writing Team Leader → {Summary Writer, Technical Writer}
                         → Compile Writing Report → CEO

4. Consolidation

CEO → Consolidation Agent → Final Report Generation → END

⚙️ Configuration

LLM Configuration (llm_config.py)

Configure your preferred LLM provider:

from langchain_groq import ChatGroq

llm = ChatGroq(
    model="mixtral-8x7b-32768",
    temperature=0.7,
    groq_api_key="your_api_key"
)

State Management (state.py)

The HierarchicalState TypedDict manages all workflow data:

class HierarchicalState(MessagesState):
    current_level: str          # Current hierarchy level
    active_teams: List[str]     # Active team identifiers
    next_agent: str            # Next agent to execute
    main_task: str             # Original user task
    research_team_data: Dict   # Research outputs
    writing_team_data: Dict    # Writing outputs
    team_completed: Dict       # Completion tracking
    final_report: str          # Consolidated output
    # ... and more

👨‍💻 Development Guide

Adding a New Agent

1. Create agent file in src/agents/:

# src/agents/my_new_agent.py
from typing import Dict
from langchain_core.messages import AIMessage
from ..core.state import HierarchicalState
from ..core.llm_config import llm

def my_new_agent(state: HierarchicalState) -> Dict:
    """Your agent description"""
    task = state.get("current_task", "")
    
    # Your agent logic here
    result = llm.invoke([HumanMessage(content=f"Process: {task}")])
    
    return {
        "messages": [AIMessage(content=f"Completed: {task}")],
        "next_agent": "appropriate_next_agent"
    }

2. Register in workflow (workflow.py):

from ..agents.my_new_agent import my_new_agent

hierarchical_wflow.add_node("my_new_agent", my_new_agent)

3. Update router (router.py):

def hierarchical_router(state: HierarchicalState) -> Literal[..., "my_new_agent", ...]:
    # Add routing logic

Adding a New Team

1. Create team leader in src/teams/
2. Add team-specific state fields in state.py
3. Update CEO decision logic in ceo.py
4. Register team leader in workflow

Modular Design Principles

• Single Responsibility: Each agent has one clear purpose
• Dependency Injection: LLM and state passed as parameters
• Type Safety: Use TypedDict for state management
• Error Handling: Graceful fallbacks in routing logic
• Testability: Pure functions for easy unit testing

🧪 Testing

Run Unit Tests

pytest tests/

Test Individual Components

# Test CEO agent
pytest tests/test_ceo.py

# Test research team
pytest tests/test_research_team.py

# Test complete workflow
pytest tests/test_workflow.py

Test Coverage

pytest --cov=src tests/

📊 Example Output

🚀 Starting Hierarchical Multi-Agent System...

📋 Task: Analyze the impact of AI on healthcare industry
============================================================

CEO - Research: False, Writing: False, Stage: planning
CEO Decision: start research
📋 CEO: Starting research phase.

Research Leader - Data: False, Market: False
📋 Research Team Leader: Assigning data research task.

Data Researcher: Completed data analysis for 'Analyze the impact of AI on healthcare industry'.
Key data points:
The healthcare AI market is projected to reach $187.95 billion by 2030...

[... workflow continues ...]

✅ CEO: Both teams completed. Moving to consolidation.

═══════════════════════════════════════════════════════
FINAL COMPREHENSIVE REPORT
═══════════════════════════════════════════════════════

[Final consolidated report appears here]

🤝 Contributing

Contributions are welcome! Please follow these guidelines:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/AmazingFeature)
3. Commit your changes (git commit -m 'Add some AmazingFeature')
4. Push to the branch (git push origin feature/AmazingFeature)
5. Open a Pull Request

Development Setup

# Install development dependencies
pip install -r requirements-dev.txt

# Run linting
flake8 src/

# Run type checking
mypy src/

# Format code
black src/

📝 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

• LangChain for LLM orchestration
• LangGraph for stateful workflow management
• Streamlit for the web interface
• Groq for fast LLM inference

📬 Contact

For questions or feedback, please open an issue on GitHub.

---

Built with ❤️ using LangGraph and LangChain