HierarchicalStructuredCommunicationFramework

Overview

The Hierarchical Structured Communication Framework implements the “Talk Structurally, Act Hierarchically” approach for LLM multi-agent systems, based on the research paper arXiv:2502.11098. It provides structured communication protocols with specialized agent classes for content generation, evaluation, refinement, and supervision.

Installation

pip install -U swarms

Key Components

Agent Classes

HierarchicalStructuredCommunicationGenerator - Creates initial content
HierarchicalStructuredCommunicationEvaluator - Evaluates content quality
HierarchicalStructuredCommunicationRefiner - Improves content based on feedback
HierarchicalStructuredCommunicationSupervisor - Coordinates workflow

Main Framework

HierarchicalStructuredCommunicationFramework - Main orchestrator class

Attributes

str

default:"HierarchicalStructuredCommunicationFramework"

Name of the framework instance

Optional[Union[Agent, Callable, Any]]

default:"None"

Main supervisor agent that coordinates the workflow. If not provided, a default Agent supervisor is created automatically.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of generator agents for creating initial content. If not provided, a single default generator agent is created automatically.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of evaluator agents for assessing content quality. If not provided (and enable_hierarchical_evaluation=True), a single default evaluator agent is created automatically.

Optional[List[Union[Agent, Callable, Any]]]

default:"None"

List of refiner agents for improving content based on feedback. If not provided, a single default refiner agent is created automatically.

Optional[Union[Agent, Callable, Any]]

default:"None"

Dedicated supervisor that coordinates the hierarchical evaluation phase. Created automatically if not provided.

int

default:"3"

Maximum number of refinement loops

OutputType

default:"dict-all-except-first"

Format applied to the conversation history.

str

default:"Supervisor"

Display name for the main supervisor agent.

str

default:"EvaluationSupervisor"

Display name for the evaluation supervisor agent.

bool

default:"True"

Enable the structured communication protocol with Message (M_ij), Background (B_ij), and Intermediate Output (I_ij)

bool

default:"True"

Enable hierarchical evaluation with supervisor coordination

bool

default:"True"

Enable shared memory between agents

str

default:"gpt-5.4"

LLM model name to use for the agents

bool

default:"False"

Enable verbose logging

bool

default:"False"

Route agent calls through a local Ollama server instead of a hosted provider.

str

default:"http://localhost:11434/v1"

Base URL for the Ollama server when use_ollama=True.

str

default:"ollama"

API key sent to the Ollama server when use_ollama=True.

Methods

run()

Execute the complete workflow for a given task, looping up to max_loops times (set at construction).

def run(self, task: str, img: str = None, *args, **kwargs) -> dict

Parameters:

task (str): The task to execute
img (str, optional): Optional image input

Returns: A dictionary containing final_result, total_loops, conversation_history, evaluation_results, and intermediate_outputs

run() does not accept a max_loops override — the number of refinement loops is fixed by the max_loops value passed to the constructor.

step()

Execute a single workflow step (generate, evaluate, refine).

def step(self, task: str, img: str = None, *args, **kwargs) -> dict

Parameters:

task (str): The task to execute for one step
img (str, optional): Optional image input

Returns: A dictionary with generator_result, evaluation_results, refined_result, and conversation_history (or an error key on failure)

send_structured_message()

Send a structured communication message between agents, following the Message (M_ij) / Background (B_ij) / Intermediate Output (I_ij) protocol.

def send_structured_message(
    self,
    sender: str,
    recipient: str,
    message: str,
    background: str = "",
    intermediate_output: str = "",
) -> StructuredMessage

Parameters:

sender (str): Name of the sending agent
recipient (str): Name of the receiving agent
message (str): Specific task message (M_ij)
background (str, optional): Background context (B_ij)
intermediate_output (str, optional): Intermediate output (I_ij)

Returns: The StructuredMessage that was appended to conversation_history

run_hierarchical_evaluation()

Run the hierarchical evaluation system with supervisor coordination.

def run_hierarchical_evaluation(
    self, content: str, evaluation_criteria: List[str] = None
) -> List[EvaluationResult]

Parameters:

content (str): Content to evaluate
evaluation_criteria (List[str], optional): Criteria to evaluate against. Defaults to ["accuracy", "completeness", "clarity", "relevance"].

Returns: A list of EvaluationResult objects, one per evaluator

Usage Examples

Quick Start

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework,
    HierarchicalStructuredCommunicationGenerator,
    HierarchicalStructuredCommunicationEvaluator,
    HierarchicalStructuredCommunicationRefiner,
    HierarchicalStructuredCommunicationSupervisor
)

# Create specialized agents
generator = HierarchicalStructuredCommunicationGenerator(
    agent_name="ContentGenerator"
)

evaluator = HierarchicalStructuredCommunicationEvaluator(
    agent_name="QualityEvaluator"
)

refiner = HierarchicalStructuredCommunicationRefiner(
    agent_name="ContentRefiner"
)

supervisor = HierarchicalStructuredCommunicationSupervisor(
    agent_name="WorkflowSupervisor"
)

# Create the framework
framework = HierarchicalStructuredCommunicationFramework(
    name="MyFramework",
    supervisor=supervisor,
    generators=[generator],
    evaluators=[evaluator],
    refiners=[refiner],
    max_loops=3
)

# Run the workflow
result = framework.run("Create a comprehensive analysis of AI trends in 2024")

Basic Usage with Default Supervisor

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework,
    HierarchicalStructuredCommunicationGenerator,
    HierarchicalStructuredCommunicationEvaluator,
    HierarchicalStructuredCommunicationRefiner
)

# Create agents with custom names
generator = HierarchicalStructuredCommunicationGenerator(agent_name="ContentGenerator")
evaluator = HierarchicalStructuredCommunicationEvaluator(agent_name="QualityEvaluator")
refiner = HierarchicalStructuredCommunicationRefiner(agent_name="ContentRefiner")

# Create framework with default supervisor
framework = HierarchicalStructuredCommunicationFramework(
    generators=[generator],
    evaluators=[evaluator],
    refiners=[refiner],
    max_loops=3,
    verbose=True
)

# Execute task
result = framework.run("Write a detailed report on renewable energy technologies")
print(result["final_result"])

Advanced Configuration

from swarms.structs.hierarchical_structured_communication_framework import (
    HierarchicalStructuredCommunicationFramework
)

# Create framework with custom configuration
framework = HierarchicalStructuredCommunicationFramework(
    name="AdvancedFramework",
    max_loops=5,
    enable_structured_communication=True,
    enable_hierarchical_evaluation=True,
    shared_memory=True,
    model_name="claude-sonnet-4-6",
    verbose=True
)

# Run the task (loops up to the max_loops set above)
result = framework.run(
    "Analyze the impact of climate change on global agriculture"
)

How It Works

The framework operates through a structured multi-phase workflow:

Generation Phase: Generator agents create initial content based on the task
Evaluation Phase: Evaluator agents assess the quality of generated content using structured communication protocols
Refinement Phase: Refiner agents improve content based on evaluation feedback
Supervision: The supervisor agent coordinates the entire workflow, deciding when to iterate or finalize
Iteration: Steps 1-4 repeat up to max_loops times until quality thresholds are met

Structured Communication Protocol

The framework uses a formal communication protocol with three components:

Message (M_ij): Direct communication between agents
Background (B_ij): Contextual information shared between agents
Intermediate Output (I_ij): Partial results passed between workflow stages

Features

Structured Communication: Formal protocol for inter-agent messaging
Hierarchical Evaluation: Multi-level quality assessment with supervisor oversight
Iterative Refinement: Content improves through generate-evaluate-refine loops
Specialized Agents: Purpose-built agent classes for each workflow role
Configurable: Flexible configuration for communication, evaluation, and memory
Shared Memory: Optional shared memory between agents for context retention

Source Code

View the source code on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

HierarchicalStructuredCommunicationFramework

Overview

Installation

Key Components

Agent Classes

Main Framework

Attributes

Methods

run()

step()

send_structured_message()

run_hierarchical_evaluation()

Usage Examples

Quick Start

Basic Usage with Default Supervisor

Advanced Configuration

How It Works

Structured Communication Protocol

Features

Source Code

​Overview

​Installation

​Key Components

​Agent Classes

​Main Framework

​Attributes

​Methods

​run()

​step()

​send_structured_message()

​run_hierarchical_evaluation()

​Usage Examples

​Quick Start

​Basic Usage with Default Supervisor

​Advanced Configuration

​How It Works

​Structured Communication Protocol

​Features

​Source Code

Overview

Installation

Key Components

Agent Classes

Main Framework

Attributes

Methods

run()

step()

send_structured_message()

run_hierarchical_evaluation()

Usage Examples

Quick Start

Basic Usage with Default Supervisor

Advanced Configuration

How It Works

Structured Communication Protocol

Features

Source Code